LangChain 1.0 & LangGraph 1.0 完整知识点手册

暂无标签

LangChain 1.0 & LangGraph 1.0 完整知识点手册

本手册整理自 LangChain1.0-Langgraph1.0-Learning 学习资料，涵盖从基础到高级的完整知识体系。

目录

  1. Phase 1: 基础知识

  2. Phase 2: 中级主题

  3. Phase 3: 高级主题

Phase 1: 基础知识

1.1 模型调用

核心概念

init_chat_model 是 LangChain 1.0 中用于初始化聊天模型的统一接口。

基本语法

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "provider:model_name",  # 提供商:模型名称
    api_key="your-api-key",  # API 密钥（可选，可从环境变量读取）
    temperature=0.7,         # 温度参数（可选）
    max_tokens=1000,         # 最大 token 数（可选）
)

支持的提供商格式

# Groq
"groq:llama-3.3-70b-versatile"
"groq:mixtral-8x7b-32768"

# OpenAI
"openai:gpt-4"
"openai:gpt-3.5-turbo"

# Anthropic
"anthropic:claude-sonnet-4-5-20250929"

invoke 方法 - 三种输入格式

格式 1：纯字符串（最简单）

response = model.invoke("你的问题或指令")
print(response.content)

格式 2：字典列表（推荐）

messages = [
    {"role": "system", "content": "系统提示"},
    {"role": "user", "content": "用户消息"},
    {"role": "assistant", "content": "AI回复"},  # 可选，用于对话历史
]
response = model.invoke(messages)

格式 3：消息对象列表（类型安全）

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="系统提示"),
    HumanMessage(content="用户消息"),
]
response = model.invoke(messages)

invoke 返回值详解

response = model.invoke("Hello")

# 主要内容
response.content              # str - AI 的回复文本
response.response_metadata    # dict - 响应元数据
response.id                   # str - 消息唯一 ID

# Token 使用情况
usage = response.response_metadata.get('token_usage', {})
print(f"提示 tokens: {usage.get('prompt_tokens')}")
print(f"完成 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")

完整示例代码

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载环境变量
load_dotenv()

# 初始化模型
model = init_chat_model(
    "groq:llama-3.3-70b-versatile",
    api_key=os.getenv("GROQ_API_KEY"),
    temperature=0.7,
    max_tokens=1000
)

# 简单调用
response = model.invoke("什么是机器学习？")
print(response.content)

# 带系统提示的调用
messages = [
    {"role": "system", "content": "你是一个专业的 Python 导师"},
    {"role": "user", "content": "什么是装饰器？"}
]
response = model.invoke(messages)
print(response.content)

1.2 提示词模板

核心概念

提示词模板将提示词结构与数据分离，提高可维护性和可复用性。

PromptTemplate - 简单文本模板

from langchain_core.prompts import PromptTemplate

# 方法 1：from_template（推荐）
template = PromptTemplate.from_template(
    "将以下文本翻译成{language}：\n{text}"
)

# 使用模板
prompt = template.format(language="法语", text="Hello, how are you?")
print(prompt)

# 方法 2：显式指定变量
template = PromptTemplate(
    input_variables=["product", "feature"],
    template="为{product}写一句广告语，重点突出{feature}特点。"
)

ChatPromptTemplate - 聊天消息模板

from langchain_core.prompts import ChatPromptTemplate

# 使用元组格式（推荐）
template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，擅长{skill}"),
    ("user", "{question}")
])

messages = template.format_messages(
    role="编程导师",
    skill="用简单语言解释复杂概念",
    question="什么是递归？"
)

response = model.invoke(messages)

部分变量（Partial Variables）

# 预填充某些固定变量
base_template = ChatPromptTemplate.from_messages([
    ("system", "你是{department}的{role}"),
    ("user", "{task}")
])

# 创建 IT 部门专用模板
it_template = base_template.partial(
    department="IT 部门",
    role="技术支持"
)

# 现在只需要提供 task
messages = it_template.format_messages(task="解决网络问题")

LCEL 链式调用

# 使用 | 运算符创建链
template = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("user", "{input}")
])

chain = template | model

# 直接调用链
response = chain.invoke({
    "role": "Python 导师",
    "input": "什么是装饰器？"
})
print(response.content)

1.3 消息类型与对话管理

三种消息类型

角色字典格式对象格式用途
System{"role": "system", ...}SystemMessage(...)系统提示
User{"role": "user", ...}HumanMessage(...)用户输入
Assistant{"role": "assistant", ...}AIMessage(...)AI 回复

核心难点：对话历史管理

关键规则：每次调用必须传递完整的对话历史！

❌ 错误做法

# 第一次
r1 = model.invoke("我叫张三")

# 第二次（没传历史）
r2 = model.invoke("我叫什么？")  # AI 不记得！

✅ 正确做法

conversation = []

# 第一次
conversation.append({"role": "user", "content": "我叫张三"})
r1 = model.invoke(conversation)

# 关键：保存 AI 回复
conversation.append({"role": "assistant", "content": r1.content})

# 第二次（传递完整历史）
conversation.append({"role": "user", "content": "我叫什么？"})
r2 = model.invoke(conversation)  # AI 记得！

对话历史优化

def keep_recent_messages(messages, max_pairs=3):
    """
    保留最近的 N 轮对话
    max_pairs: 保留的对话轮数（每轮 = user + assistant）
    """
    # 分离 system 和对话
    system_msgs = [m for m in messages if m.get("role") == "system"]
    conversation = [m for m in messages if m.get("role") != "system"]
    
    # 只保留最近的
    recent = conversation[-(max_pairs * 2):]
    
    # 返回：system + 最近对话
    return system_msgs + recent

1.4 自定义工具

@tool 装饰器

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气信息

    参数:
        city: 城市名称，如"北京"、"上海"

    返回:
        天气信息字符串
    """
    # 你的实现
    return "晴天，温度 15°C"

关键要点

必需项说明
@tool 装饰器声明这是一个工具
docstringAI 读这个来理解工具用途 ⚠️ 非常重要！
类型注解参数和返回值的类型
返回 str工具应该返回字符串

多参数工具

@tool
def calculator(operation: str, a: float, b: float) -> str:
    """
    执行数学计算

    参数:
        operation: "add", "subtract", "multiply", "divide"
        a: 第一个数字
        b: 第二个数字
    """
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else "错误：除数不能为零"
    }
    result = operations.get(operation, lambda x, y: "未知操作")(a, b)
    return f"{a} {operation} {b} = {result}"

可选参数工具

from typing import Optional

@tool
def web_search(query: str, num_results: Optional[int] = 3) -> str:
    """
    搜索网页

    参数:
        query: 搜索关键词
        num_results: 返回结果数量，默认 3
    """
    # 实现搜索逻辑
    return f"找到 {num_results} 条关于 '{query}' 的结果"

工具绑定到模型

# 绑定工具到模型
model_with_tools = model.bind_tools([get_weather, calculator])

# AI 可以决定是否调用工具
response = model_with_tools.invoke("北京天气如何？")

# 检查 AI 是否要调用工具
if response.tool_calls:
    print("AI 想调用工具：", response.tool_calls)
else:
    print("AI 直接回答：", response.content)

1.5 Agent 创建

核心概念

Agent = 模型 + 工具 + 自动决策

create_agent 基本用法

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

agent = create_agent(
    model=init_chat_model("groq:llama-3.3-70b-versatile"),
    tools=[tool1, tool2],
    system_prompt="Agent 的行为指令"  # 可选
)

response = agent.invoke({
    "messages": [{"role": "user", "content": "问题"}]
})

# 获取最终回答
final_answer = response['messages'][-1].content

Agent 执行循环

用户问题 (HumanMessage)
    ↓
AI 决定 (AIMessage with tool_calls)
    ↓
执行工具 (ToolMessage)
    ↓
最终答案 (AIMessage)

多轮对话 Agent

from langgraph.checkpoint.memory import MemorySaver

# 创建带记忆的 Agent
agent = create_agent(
    model=model,
    tools=[calculator],
    system_prompt="你是一个有帮助的助手。",
    checkpointer=MemorySaver()  # 添加内存
)

# 使用 thread_id 来保持对话
config = {"configurable": {"thread_id": "conversation-1"}}

# 第一轮
response1 = agent.invoke(
    {"messages": [{"role": "user", "content": "10 加 5 等于多少？"}]},
    config=config
)

# 第二轮（记忆自动保持）
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "再乘以 3 呢？"}]},
    config=config
)

1.6 Agent 执行循环

查看完整执行过程

response = agent.invoke({
    "messages": [{"role": "user", "content": "25 乘以 8 等于多少？"}]
})

# 显示完整的消息历史
for i, msg in enumerate(response['messages'], 1):
    print(f"\n--- 消息 {i} ({msg.__class__.__name__}) ---")
    if hasattr(msg, 'content'):
        print(f"内容：{msg.content}")
    if hasattr(msg, 'tool_calls') and msg.tool_calls:
        print(f"工具调用：{msg.tool_calls}")

流式输出

# 使用 .stream() 实时输出
for chunk in agent.stream({"messages": [{"role": "user", "content": "问题"}]}):
    # 实时处理每个步骤
    print(chunk)

Phase 2: 中级主题

2.1 内存管理

核心概念

内存 = Agent 记住对话历史的能力

没有内存（默认）

agent = create_agent(model=model, tools=[])

# 第一轮
agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]})

# 第二轮 - 不记得第一轮！
response = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]})
# AI 会说"不知道"

添加内存

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

# 创建 Agent 时添加 checkpointer
agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver()  # 添加内存
)

# 调用时指定 thread_id
config = {"configurable": {"thread_id": "conversation_1"}}

# 第一轮
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config=config
)

# 第二轮 - 记得第一轮！
response = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config
)
# AI 会说"你叫张三"

多会话管理

# 会话 1 - Alice
config_alice = {"configurable": {"thread_id": "user_alice"}}
agent.invoke({"messages": [...]}, config_alice)

# 会话 2 - Bob（完全独立）
config_bob = {"configurable": {"thread_id": "user_bob"}}
agent.invoke({"messages": [...]}, config_bob)

2.2 上下文管理

问题：对话历史会无限增长

  • 消息越来越多
  • 超过模型的 token 限制
  • 响应变慢、成本增加

解决方案：修剪对话历史

def trim_conversation(messages, max_tokens=4000):
    """修剪对话历史，保持在 token 限制内"""
    # 保留 system 消息
    system_msgs = [m for m in messages if m.get("role") == "system"]
    conversation = [m for m in messages if m.get("role") != "system"]
    
    # 从最新的消息开始保留
    trimmed = []
    current_tokens = 0
    
    for msg in reversed(conversation):
        msg_tokens = len(msg.get("content", "")) // 4  # 粗略估算
        if current_tokens + msg_tokens > max_tokens:
            break
        trimmed.insert(0, msg)
        current_tokens += msg_tokens
    
    return system_msgs + trimmed

2.3 检查点持久化

InMemorySaver 的限制

  • ✅ 同一进程内有效
  • ❌ 程序重启后丢失
  • ❌ 不同进程无法共享

SQLite 持久化（后续学习）

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 持久化
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    agent = create_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer
    )

2.4 结构化输出

使用 Pydantic 定义输出结构

from pydantic import BaseModel, Field
from typing import List

class PersonInfo(BaseModel):
    """人物信息结构"""
    name: str = Field(description="人物姓名")
    age: int = Field(description="年龄")
    skills: List[str] = Field(description="技能列表")

# 让模型输出结构化数据
model_with_structure = model.with_structured_output(PersonInfo)

result = model_with_structure.invoke("介绍一下 Elon Musk")
print(result.name)      # "Elon Musk"
print(result.age)       # 52
print(result.skills)    # ["创业", "工程", ...]

2.5 RAG 基础

核心概念

RAG (Retrieval-Augmented Generation) = 检索增强生成

RAG 工作流程：

1. 离线：文档 → 分割 → 嵌入 → 存入向量数据库
2. 在线：用户查询 → 检索相关文档 → 提供给 LLM → 生成答案

完整实现

# 1. 文档加载
from langchain_community.document_loaders import TextLoader

loader = TextLoader("document.txt", encoding="utf-8")
documents = loader.load()

# 2. 文本分割
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大字符数
    chunk_overlap=50,      # 块之间的重叠
)
chunks = splitter.split_documents(documents)

# 3. 向量嵌入
from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"  # 免费模型
)

# 4. 存入 Pinecone
from langchain_pinecone import PineconeVectorStore

vectorstore = PineconeVectorStore.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="my-index"
)

# 5. 检索
docs = vectorstore.similarity_search("查询文本", k=3)

# 6. 创建 RAG Agent
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search_kb(query: str) -> str:
    """搜索知识库"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([doc.page_content for doc in docs])

agent = create_agent(
    model=model,
    tools=[search_kb],
    system_prompt="使用 search_kb 工具检索信息，然后回答问题。"
)

2.6 RAG 进阶

混合搜索

# 向量搜索 + 关键词搜索
results = vectorstore.similarity_search(
    query="查询",
    k=3,
    filter={"category": "tech"}  # 元数据过滤
)

重排序（Reranking）

# 先检索更多结果，然后重排序
initial_results = vectorstore.similarity_search(query, k=10)
reranked_results = reranker.rerank(query, initial_results, top_k=3)

Phase 3: 高级主题

3.1 LangGraph 基础

核心概念

LangGraph 使用图（Graph） 来组织工作流：

  • 节点（Nodes）：处理单元
  • 边（Edges）：连接节点的路径
  • 状态（State）：在节点之间传递的数据

基本用法

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表，自动累加
    current_step: str

# 2. 定义节点
def process_node(state: State) -> dict:
    # 处理逻辑
    return {"current_step": "completed"}

# 3. 创建图
graph = StateGraph(State)
graph.add_node("process", process_node)
graph.add_edge(START, "process")
graph.add_edge("process", END)

# 4. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [], "current_step": "start"})

条件边

def route_function(state: State) -> str:
    """根据状态决定下一个节点"""
    if state["current_step"] == "need_research":
        return "researcher"
    else:
        return "writer"

graph.add_conditional_edges(
    "supervisor",
    route_function,
    {
        "researcher": "researcher_node",
        "writer": "writer_node"
    }
)

3.2 多 Agent 协作

监督者模式（Supervisor Pattern）

        ┌──────────────┐
        │  Supervisor  │
        │   (协调者)    │
        └──────┬───────┘
   ┌───────────┼───────────┐
   ▼           ▼           ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent C │
│(研究员) │ │ (编辑)  │ │(审核员) │
└─────────┘ └─────────┘ └─────────┘

完整实现

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 定义状态
class TeamState(TypedDict):
    task: str
    messages: Annotated[list, add_messages]
    research_result: str
    draft: str
    final_content: str
    next_agent: str

# 监督者节点
def supervisor(state: TeamState) -> dict:
    """决定下一步由哪个 Agent 处理"""
    if not state.get("research_result"):
        next_agent = "researcher"
    elif not state.get("draft"):
        next_agent = "writer"
    elif not state.get("final_content"):
        next_agent = "editor"
    else:
        next_agent = "complete"
    
    return {"next_agent": next_agent}

# 研究员 Agent
def researcher(state: TeamState) -> dict:
    """收集和整理信息"""
    # 使用 LLM 进行研究
    response = model.invoke([
        {"role": "system", "content": "你是一个研究员"},
        {"role": "user", "content": state["task"]}
    ])
    return {"research_result": response.content}

# 作家 Agent
def writer(state: TeamState) -> dict:
    """撰写内容"""
    response = model.invoke([
        {"role": "system", "content": "你是一个作家"},
        {"role": "user", "content": f"根据研究资料写作：{state['research_result']}"}
    ])
    return {"draft": response.content}

# 编辑 Agent
def editor(state: TeamState) -> dict:
    """审核和优化"""
    response = model.invoke([
        {"role": "system", "content": "你是一个编辑"},
        {"role": "user", "content": f"审核优化：{state['draft']}"}
    ])
    return {"final_content": response.content}

# 路由函数
def route_to_agent(state: TeamState) -> Literal["researcher", "writer", "editor", "complete"]:
    return state["next_agent"]

# 构建图
graph = StateGraph(TeamState)

# 添加节点
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("editor", editor)

# 添加边
graph.add_edge(START, "supervisor")

graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "writer": "writer",
        "editor": "editor",
        "complete": END
    }
)

graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("editor", "supervisor")

# 编译并运行
app = graph.compile()
result = app.invoke({"task": "写一篇关于 AI 的文章", "messages": []})

3.3 条件路由

动态分支

def classifier(state: State) -> dict:
    """分类器：识别任务类型"""
    response = model.invoke([
        {"role": "system", "content": "分类任务：billing/technical/general"},
        {"role": "user", "content": state["query"]}
    ])
    return {"category": response.content.strip().lower()}

def route_to_specialist(state: State) -> Literal["billing", "technical", "general"]:
    return state["category"]

graph.add_conditional_edges(
    "classifier",
    route_to_specialist,
    {
        "billing": "billing_agent",
        "technical": "technical_agent",
        "general": "general_agent"
    }
)

3.4 多模态处理

图像输入

from langchain_core.messages import HumanMessage

# 传递图像 URL
message = HumanMessage(content=[
    {"type": "text", "text": "这张图片里有什么？"},
    {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}}
])

response = model.invoke([message])

文件处理

from langchain_community.document_loaders import PyPDFLoader

# 加载 PDF
loader = PyPDFLoader("document.pdf")
pages = loader.load()

# 处理每一页
for page in pages:
    print(page.page_content)

3.5 LangSmith 集成

配置

# .env
LANGSMITH_API_KEY=your-key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-project

追踪和监控

# 自动追踪所有 LLM 调用
response = model.invoke("Hello")

# 在 LangSmith Dashboard 查看：
# - Token 使用
# - 延迟
# - 成本
# - 调用链

3.6 错误处理

重试机制

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def call_with_retry(messages):
    return model.invoke(messages)

降级处理

def safe_invoke(messages, fallback_model=None):
    """安全的模型调用，带降级"""
    try:
        return model.invoke(messages)
    except Exception as e:
        print(f"主模型失败: {e}")
        if fallback_model:
            return fallback_model.invoke(messages)
        return AIMessage(content="抱歉，服务暂时不可用")

附录：快速参考

常用导入

# 模型
from langchain.chat_models import init_chat_model

# 提示词
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# 消息
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 工具
from langchain_core.tools import tool

# Agent
from langchain.agents import create_agent

# 内存
from langgraph.checkpoint.memory import InMemorySaver

# LangGraph
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 文档加载
from langchain_community.document_loaders import TextLoader, PyPDFLoader

# 文本分割
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 嵌入
from langchain_huggingface import HuggingFaceEmbeddings

# 向量存储
from langchain_pinecone import PineconeVectorStore

环境配置

# .env 文件
GROQ_API_KEY=your-groq-key
OPENAI_API_KEY=your-openai-key
PINECONE_API_KEY=your-pinecone-key
LANGSMITH_API_KEY=your-langsmith-key

安装依赖

pip install langchain langchain-groq langchain-openai langgraph langsmith
pip install langchain-huggingface langchain-pinecone pinecone-client
pip install langchain-text-splitters langchain-community
pip install python-dotenv pydantic

本手册整理自 LangChain1.0-Langgraph1.0-Learning 学习资料

标签：
暂无标签