LangGraph Chain 详细解读

📚 概述

本文档详细解读 LangGraph 的 Chain（链式调用） 基础概念。这是 LangGraph 开发的第一课，涵盖了构建智能对话系统的核心组件：消息系统、聊天模型、工具绑定和工具执行。通过本教程，你将掌握如何使用 LangChain 的基础概念构建一个简单但功能完整的 AI 链。

🎯 核心概念

什么是 Chain？

在 LangChain/LangGraph 中，Chain 是一系列按顺序执行的操作组合。一个简单的 Chain 通常包含：

1. 输入处理：接收用户消息
2. 模型调用：使用 LLM 生成响应
3. 工具执行（可选）：调用外部工具/函数
4. 输出返回：返回最终结果

本教程的四大核心概念

1. 消息（Messages）：使用聊天消息作为图状态
2. 聊天模型（Chat Models）：在图节点中使用 LLM
3. 工具绑定（Tool Binding）：为聊天模型绑定工具
4. 工具执行（Tool Calling）：在图节点中执行工具调用

系统架构图

用户输入 "Multiply 2 and 3"
        ↓
   [tool_calling_llm]
        ↓
   检测到需要调用工具
        ↓
   返回工具调用信息
   (工具名: multiply, 参数: {a:2, b:3})

---

🔧 代码实现详解

1. 消息系统（Messages）

什么是消息？

聊天模型使用 消息（Messages） 来表示对话中的不同角色。LangChain 支持多种消息类型：

消息类型	角色	用途
HumanMessage	用户	表示用户输入
AIMessage	AI 模型	表示模型响应
SystemMessage	系统	指导模型行为
ToolMessage	工具	工具调用的结果

消息的组成部分

每个消息可以包含：

• content（必需）：消息内容
• name（可选）：消息发送者的名字
• response_metadata（可选）：元数据字典（如模型响应信息）

代码示例

from langchain_core.messages import AIMessage, HumanMessage

# 创建消息列表
messages = [
    AIMessage(content="So you said you were researching ocean mammals?", name="Model"),
    HumanMessage(content="Yes, that's right.", name="Lance"),
    AIMessage(content="Great, what would you like to learn about.", name="Model"),
    HumanMessage(content="I want to learn about the best place to see Orcas in the US.", name="Lance")
]

# 打印消息
for m in messages:
    m.pretty_print()

输出：

================================== Ai Message ==================================
Name: Model
So you said you were researching ocean mammals?

================================ Human Message =================================
Name: Lance
Yes, that's right.

...

Python 知识点：for 循环与方法调用

for m in messages:
    m.pretty_print()

• for m in messages：遍历 messages 列表中的每个元素
• m.pretty_print()：调用每个消息对象的 pretty_print() 方法
• 这是 Python 面向对象编程的典型用法

---

2. 聊天模型（Chat Models）

什么是聊天模型？

聊天模型 是支持对话式交互的 LLM，可以：

• 接收消息列表作为输入
• 理解对话上下文
• 生成符合角色的回复

LangChain 支持多种聊天模型提供商：OpenAI、Anthropic、Google、Ollama 等。

设置 API Key

import os, getpass

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")

Python 知识点：环境变量与安全性

• os.environ.get(var)：获取环境变量
• getpass.getpass()：安全地输入密码（输入不会显示）
• 这是保护 API Key 的最佳实践

使用聊天模型

from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(model="gpt-4o")

# 调用模型
result = llm.invoke(messages)

输出类型：

type(result)
# langchain_core.messages.ai.AIMessage

result 是一个 AIMessage 对象，包含：

result.content  # AI 生成的文本内容
result.response_metadata  # 响应元数据

响应元数据详解

result.response_metadata

输出示例：

{
    'token_usage': {
        'completion_tokens': 228,  # 输出的 token 数
        'prompt_tokens': 67,       # 输入的 token 数
        'total_tokens': 295        # 总 token 数
    },
    'model_name': 'gpt-4o-2024-08-06',
    'finish_reason': 'stop',       # 停止原因（正常结束）
    'logprobs': None
}

重要性：

• 可以跟踪 API 使用成本
• 了解模型生成的详细信息
• 调试和优化性能

---

3. 工具（Tools）

什么是工具？

工具（Tools） 允许 LLM 与外部系统交互。当你需要：

• 调用 API
• 执行计算
• 查询数据库
• 访问外部服务

你可以将这些功能定义为"工具"，让模型知道何时以及如何使用它们。

工具的工作原理

用户输入（自然语言）
        ↓
   LLM 分析输入
        ↓
   决定是否需要调用工具
        ↓
   返回符合工具 schema 的调用信息
   (包含工具名称和参数)

定义工具

在 LangChain 中，任何 Python 函数都可以作为工具！

def multiply(a: int, b: int) -> int:
    """Multiply a and b.

    Args:
        a: first int
        b: second int
    """
    return a * b

关键要点：

1. 类型注解：a: int, b: int -> int 让模型知道参数类型
2. 文档字符串：描述函数的功能，帮助模型理解何时使用

Python 知识点：类型注解（Type Hints）

def multiply(a: int, b: int) -> int:
    #          ^^^^^^  ^^^^^^    ^^^
    #          参数a   参数b     返回值
    #          是int   是int     是int
    return a * b

类型注解的作用：

• 提供 IDE 代码提示
• 帮助 LangChain 生成工具 schema
• 提高代码可读性
• 不强制执行（Python 仍然是动态类型语言）

绑定工具到模型

llm_with_tools = llm.bind_tools([multiply])

发生了什么？

• bind_tools() 告诉模型："你可以调用这些工具"
• LangChain 自动从函数签名生成工具 schema
• 模型现在知道 multiply 工具的存在、参数和用途

工具调用示例

tool_call = llm_with_tools.invoke([
    HumanMessage(content="What is 2 multiplied by 3", name="Lance")
])

tool_call.tool_calls

输出：

[{
    'name': 'multiply',              # 工具名称
    'args': {'a': 2, 'b': 3},       # 工具参数
    'id': 'call_lBBBNo5oYpHGRqwxNaNRbsiT',  # 调用 ID
    'type': 'tool_call'             # 类型
}]

重要理解：

• LLM 并没有真正执行 multiply(2, 3)
• 它只是 返回了调用信息
• 实际执行需要在后续步骤中处理

---

4. 使用消息作为状态（Messages as State）

为什么使用消息作为状态？

在对话系统中，我们需要：

• 保存对话历史
• 追踪多轮交互
• 传递上下文信息

使用 消息列表 作为图状态是最自然的方式。

定义状态

方法 1：手动定义

from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage

class MessagesState(TypedDict):
    messages: list[AnyMessage]

Python 知识点：TypedDict

TypedDict 是 Python 的类型注解工具，用于定义字典的结构：

class MessagesState(TypedDict):
    messages: list[AnyMessage]
    #         ^^^^^^^^^^^^^^^^
    #         值的类型：AnyMessage 对象的列表

# 使用
state = {"messages": [HumanMessage(content="Hi")]}

优势：

• 提供 IDE 类型检查和自动补全
• 清晰地定义数据结构
• 不影响运行时性能（仅用于静态分析）

---

5. Reducers（归约器）⭐

问题：状态覆盖

默认情况下，节点返回的新值会 覆盖 旧值：

# 初始状态
state = {"messages": [message1, message2]}

# 节点返回
node_output = {"messages": [message3]}

# 结果：message1 和 message2 被丢弃！
new_state = {"messages": [message3]}

这不是我们想要的！ 我们希望 追加 消息，而不是覆盖。

解决方案：add_messages Reducer

Reducer 是一个函数，指定如何更新状态。

from typing import Annotated
from langgraph.graph.message import add_messages

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    #         ^^^^^^^^^                   ^^^^^^^^^^^^^
    #         基础类型                    Reducer 函数

Python 知识点：Annotated

Annotated 是 Python 3.9+ 引入的类型注解增强工具：

from typing import Annotated

Annotated[类型, 元数据1, 元数据2, ...]
#         ^^^^  ^^^^^^^^^^^^^^^^^^^^^^^^
#         基础  额外信息（不影响类型检查）

在 LangGraph 中：

messages: Annotated[list[AnyMessage], add_messages]
#                   ^^^^^^^^^^^^^^^^^  ^^^^^^^^^^^^^
#                   类型：消息列表      元数据：reducer 函数

LangGraph 会读取这个元数据，知道应该使用 add_messages 来更新状态。

add_messages 的工作原理

from langgraph.graph.message import add_messages

# 初始消息
initial = [
    AIMessage(content="Hello! How can I assist you?", name="Model"),
    HumanMessage(content="I'm looking for information on marine biology.", name="Lance")
]

# 新消息
new_message = AIMessage(content="Sure, I can help with that.", name="Model")

# 使用 add_messages
result = add_messages(initial, new_message)
# 结果：[message1, message2, message3]  ← 追加，不覆盖！

add_messages 的特性：

1. 自动追加新消息到列表末尾
2. 支持单个消息或消息列表
3. 处理消息去重（基于 ID）
4. 是 LangGraph 内置的 reducer

---

6. 使用 MessagesState（推荐方式）

由于"消息列表"是如此常见的状态结构，LangGraph 提供了预构建的 MessagesState：

from langgraph.graph import MessagesState

class MessagesState(MessagesState):
    # 如果需要额外字段，在这里添加
    pass

预构建的 MessagesState 包含：

• messages 字段：list[AnyMessage]
• 自动使用 add_messages reducer
• 无需手动定义

如果需要额外字段：

class ExtendedState(MessagesState):
    user_id: str
    session_id: str
    # messages 字段已经自动包含

---

7. 构建图

现在我们有了所有组件，可以构建图了！

from langgraph.graph import StateGraph, START, END

# 定义节点
def tool_calling_llm(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_edge(START, "tool_calling_llm")
builder.add_edge("tool_calling_llm", END)
graph = builder.compile()

代码详解

1. 定义节点函数

def tool_calling_llm(state: MessagesState):
    #                  ^^^^^^^^^^^^^^^^^^^^^^
    #                  输入：当前图状态

    result = llm_with_tools.invoke(state["messages"])
    #        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #        调用 LLM，传入对话历史

    return {"messages": [result]}
    #      ^^^^^^^^^^^^^^^^^^^^^^^
    #      返回新消息（会被 add_messages 追加）

2. 创建图

builder = StateGraph(MessagesState)
#                    ^^^^^^^^^^^^^^
#                    指定状态类型

3. 添加节点

builder.add_node("tool_calling_llm", tool_calling_llm)
#                ^^^^^^^^^^^^^^^^^^  ^^^^^^^^^^^^^^^^^
#                节点名称            节点函数

4. 添加边

builder.add_edge(START, "tool_calling_llm")
#                ^^^^^  ^^^^^^^^^^^^^^^^^^
#                起点    目标节点

builder.add_edge("tool_calling_llm", END)
#                ^^^^^^^^^^^^^^^^^^  ^^^
#                源节点              终点

LangGraph 知识点：START 和 END

• START：特殊节点，表示图的入口
• END：特殊节点，表示图的出口
• 不是字符串，而是常量标识符

5. 编译图

graph = builder.compile()

编译后的图可以执行（invoke、stream 等）。

图的可视化

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

输出：

    ┌─────────┐
    │ __start__ │
    └─────┬─────┘
          │
          ↓
┌─────────────────┐
│ tool_calling_llm │
└─────────┬─────────┘
          │
          ↓
    ┌─────────┐
    │ __end__  │
    └─────────┘

---

8. 执行图

场景 1：普通对话

messages = graph.invoke({"messages": HumanMessage(content="Hello!")})

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Hello!

================================== Ai Message ==================================
Hi there! How can I assist you today?

分析：

• 输入不需要工具，LLM 直接回复
• 没有工具调用信息

场景 2：工具调用

messages = graph.invoke({"messages": HumanMessage(content="Multiply 2 and 3")})

for m in messages['messages']:
    m.pretty_print()

输出：

================================ Human Message =================================
Multiply 2 and 3

================================== Ai Message ==================================
Tool Calls:
  multiply (call_Er4gChFoSGzU7lsuaGzfSGTQ)
 Call ID: call_Er4gChFoSGzU7lsuaGzfSGTQ
  Args:
    a: 2
    b: 3

分析：

• LLM 识别出需要使用 multiply 工具
• 返回工具调用信息，但 尚未执行
• 参数 a=2, b=3 从自然语言中提取

---

🎓 核心知识点总结

LangGraph/LangChain 特有概念

1. 消息系统

消息类型	用途	示例
HumanMessage	用户输入	HumanMessage(content="Hi")
AIMessage	模型输出	AIMessage(content="Hello!")
SystemMessage	系统指令	SystemMessage(content="You are helpful")
ToolMessage	工具结果	ToolMessage(content="6", tool_call_id="...")

2. 工具绑定

# 定义工具（普通 Python 函数）
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

# 绑定到模型
llm_with_tools = llm.bind_tools([multiply])

关键点：

• 函数签名 → 自动生成工具 schema
• 文档字符串 → 帮助模型理解用途
• 类型注解 → 定义参数和返回值类型

3. Reducers

作用： 控制状态更新方式

# 默认行为：覆盖
messages: list[AnyMessage]

# 使用 reducer：追加
messages: Annotated[list[AnyMessage], add_messages]

常用 reducers：

• add_messages：追加消息（去重）
• operator.add：列表拼接
• 自定义函数：完全控制更新逻辑

4. 图的构建模式

# 1. 创建图
builder = StateGraph(StateClass)

# 2. 添加节点
builder.add_node("node_name", node_function)

# 3. 添加边
builder.add_edge(START, "node_name")
builder.add_edge("node_name", END)

# 4. 编译
graph = builder.compile()

# 5. 执行
result = graph.invoke(initial_state)

Python 特有知识点

1. TypedDict

作用： 定义字典的结构和类型

from typing_extensions import TypedDict

class MyState(TypedDict):
    name: str
    age: int

# 使用
state: MyState = {"name": "Alice", "age": 30}

特点：

• 仅用于类型检查（静态分析）
• 运行时仍是普通字典
• 提供 IDE 自动补全

2. Type Hints（类型注解）

def greet(name: str) -> str:
    #         ^^^^^    ^^^^^
    #         参数类型  返回值类型
    return f"Hello, {name}"

用途：

• 文档化代码
• IDE 支持
• 静态类型检查（mypy）
• LangChain 工具 schema 生成

3. Annotated

作用： 为类型添加元数据

from typing import Annotated

# 基础类型 + 元数据
Age = Annotated[int, "Must be positive"]

# 在 LangGraph 中
messages: Annotated[list, add_messages]
#                   ^^^^  ^^^^^^^^^^^^^
#                   类型  元数据（reducer）

LangGraph 使用场景：

class State(TypedDict):
    # 会被覆盖
    count: int

    # 会被累加
    scores: Annotated[list[int], operator.add]

    # 会被追加（去重）
    messages: Annotated[list[AnyMessage], add_messages]

4. 环境变量

import os

# 设置
os.environ["KEY"] = "value"

# 获取
value = os.environ.get("KEY")

# 安全输入
import getpass
api_key = getpass.getpass("API Key: ")

---

💡 最佳实践

1. 消息管理

✅ 好的做法

# 明确指定消息角色和名称
messages = [
    SystemMessage(content="You are a helpful assistant"),
    HumanMessage(content="What's the weather?", name="User"),
    AIMessage(content="I'll check that for you", name="Assistant")
]

❌ 避免的做法

# 不要混用字符串和消息对象
messages = ["Hello", HumanMessage(content="Hi")]  # 错误！

2. 工具定义

✅ 好的做法

def calculate_discount(price: float, discount_percent: float) -> float:
    """Calculate the discounted price.

    Args:
        price: Original price in dollars
        discount_percent: Discount percentage (0-100)

    Returns:
        Final price after discount
    """
    return price * (1 - discount_percent / 100)

特点：

• 完整的类型注解
• 详细的文档字符串
• 清晰的参数说明

❌ 避免的做法

def calc(p, d):  # 无类型注解，无文档
    return p * (1 - d / 100)

3. 状态设计

✅ 好的做法

class ChatState(MessagesState):
    user_id: str
    session_started_at: str
    # messages 自动继承，带 add_messages reducer

❌ 避免的做法

# 不要手动管理消息列表
class ChatState(TypedDict):
    all_messages: list  # 无类型，无 reducer

4. 节点函数

✅ 好的做法

def process_message(state: MessagesState) -> dict:
    """Process user message and generate response."""
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        logger.error(f"LLM error: {e}")
        return {"messages": [AIMessage(content="Sorry, an error occurred")]}

特点：

• 明确的类型注解
• 错误处理
• 返回正确的状态格式

❌ 避免的做法

def process(s):  # 无类型，无错误处理
    return llm.invoke(s["messages"])  # 返回格式错误

---

🚀 进阶技巧

1. 自定义 Reducer

除了 add_messages，你可以定义自定义 reducer：

def keep_last_n_messages(existing: list, new: list, n: int = 10) -> list:
    """只保留最近 N 条消息"""
    combined = existing + new
    return combined[-n:]

class State(TypedDict):
    messages: Annotated[list, lambda x, y: keep_last_n_messages(x, y, 5)]

2. 条件工具调用

def smart_node(state: MessagesState):
    response = llm_with_tools.invoke(state["messages"])

    if response.tool_calls:
        # 有工具调用 → 执行工具
        tool_name = response.tool_calls[0]["name"]
        tool_args = response.tool_calls[0]["args"]

        if tool_name == "multiply":
            result = multiply(**tool_args)
            return {"messages": [
                response,
                ToolMessage(content=str(result), tool_call_id=response.tool_calls[0]["id"])
            ]}

    # 无工具调用 → 直接返回
    return {"messages": [response]}

3. 多模型支持

class ModelState(MessagesState):
    current_model: str

def adaptive_node(state: ModelState):
    # 根据状态选择不同模型
    if state["current_model"] == "fast":
        llm = ChatOpenAI(model="gpt-3.5-turbo")
    else:
        llm = ChatOpenAI(model="gpt-4o")

    response = llm.invoke(state["messages"])
    return {"messages": [response]}

---

🔍 常见问题

Q1: 为什么需要 add_messages reducer？

答： 因为对话需要保留历史记录。

# 没有 reducer（错误）
state = {"messages": [msg1, msg2]}
node_output = {"messages": [msg3]}
# 结果：msg1, msg2 丢失！

# 有 add_messages（正确）
state = {"messages": [msg1, msg2]}
node_output = {"messages": [msg3]}
# 结果：[msg1, msg2, msg3]  ← 保留历史

Q2: invoke() 和 stream() 有什么区别？

# invoke：等待全部完成，返回最终结果
result = graph.invoke({"messages": [HumanMessage(content="Hi")]})

# stream：流式返回中间结果
for chunk in graph.stream({"messages": [HumanMessage(content="Hi")]}):
    print(chunk)

使用场景：

• invoke：需要完整结果时
• stream：需要实时反馈时（如聊天界面）

Q3: 工具调用后如何执行工具？

本教程中，LLM 只返回工具调用信息，不执行。完整的工具执行流程需要：

1. 检测 response.tool_calls
2. 调用实际函数
3. 将结果作为 ToolMessage 返回
4. 再次调用 LLM 生成最终回答

这将在后续教程中详细介绍。

Q4: TypedDict 和 Pydantic BaseModel 如何选择？

场景	使用
定义图状态	TypedDict
LLM 结构化输出	BaseModel
API 请求/响应	BaseModel
需要数据验证	BaseModel
仅需类型提示	TypedDict

# 图状态
class State(TypedDict):
    messages: list

# 工具输出
class ToolOutput(BaseModel):
    result: int
    status: str

---

📊 概念对比

Chain vs Graph

特性	Chain	Graph
结构	线性（A→B→C）	任意（可循环、分支）
复杂度	简单	复杂
适用场景	简单流程	复杂工作流
本教程	✅	✅（简单图）

本教程使用了简单的线性图（chain），但使用了 StateGraph 构建，为后续复杂图打基础。

Messages vs State

# 仅消息
def node1(messages: list[AnyMessage]) -> list[AnyMessage]:
    return llm.invoke(messages)

# 使用状态（推荐）
def node2(state: MessagesState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

使用状态的优势：

• 可以添加额外信息（user_id、metadata 等）
• 统一的更新机制（reducers）
• 更好的扩展性

---

🎯 实际应用案例

案例 1：简单客服机器人

from langgraph.graph import StateGraph, MessagesState, START, END

# 工具：查询订单
def check_order(order_id: str) -> str:
    """Check order status"""
    return f"Order {order_id} is being processed"

# 节点
def customer_service(state: MessagesState):
    llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools([check_order])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("service", customer_service)
builder.add_edge(START, "service")
builder.add_edge("service", END)
app = builder.compile()

# 使用
result = app.invoke({"messages": [HumanMessage(content="Check order #12345")]})

案例 2：多语言翻译助手

def translate(text: str, target_language: str) -> str:
    """Translate text to target language"""
    # 实际调用翻译 API
    return f"[{target_language}] {text}"

def translator_bot(state: MessagesState):
    llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools([translate])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 使用
result = app.invoke({
    "messages": [HumanMessage(content="Translate 'Hello' to Spanish")]
})

---

📖 下一步学习

本教程介绍了 Chain 的基础概念，下一步你将学习：

1. Tool Execution（工具执行）：实际执行工具调用并返回结果
2. Conditional Edges（条件边）：根据状态动态路由
3. Cycles（循环）：构建可以多轮交互的图
4. Memory（记忆）：持久化对话历史

---

🔗 扩展阅读

• LangChain Messages 文档
• LangChain Chat Models 文档
• LangChain Tools 文档
• LangGraph Reducers 文档
• Python Type Hints 指南

---

总结：本教程介绍了构建 LangGraph Chain 的四大核心概念：消息系统、聊天模型、工具绑定和状态管理。这些是构建复杂 AI 系统的基础，掌握这些概念后，你就可以开始构建更强大的 LangGraph 应用了！