小结和复习: LangGraph 框架基础

复习寄语
恭喜你完成了 Module 1 的学习!作为一位教育者,我深知主动回顾是巩固知识的最有效方法。本章将通过 15 个精心设计的问题,帮助你检验理解深度,发现知识盲点,并建立系统性的认知框架。

📚 术语表

术语名称	LangGraph 定义和解读	Python 定义和说明	重要程度
Graph State Machine	图状态机，LangGraph 的核心架构，将 Agent 建模为有向图	设计模式，通过节点和边定义状态转换和执行流程	⭐⭐⭐⭐⭐
StateGraph	状态图类，用于构建和管理 LangGraph 工作流	Python 类，提供 add_node/add_edge/compile 等方法	⭐⭐⭐⭐⭐
State Schema	状态模式，定义图中流转的数据结构	TypedDict 或 Pydantic BaseModel，定义字段和类型	⭐⭐⭐⭐⭐
Node	节点函数，接收状态并返回状态更新的执行单元	Python 函数，签名为 def node(state: State) -> dict	⭐⭐⭐⭐⭐
Edge	边，定义节点间的连接和执行顺序	add_edge(from_node, to_node) 方法定义的路径	⭐⭐⭐⭐⭐
Conditional Edge	条件边，根据状态动态路由到不同节点	add_conditional_edges() 配合路由函数实现分支逻辑	⭐⭐⭐⭐⭐
Checkpointer	检查点保存器，持久化图的执行状态	MemorySaver、SqliteSaver 等类，支持状态恢复	⭐⭐⭐⭐
Reducer	归约器，定义如何合并状态更新	add_messages 等函数，控制状态字段的更新逻辑	⭐⭐⭐⭐
Breakpoint	断点，在特定节点暂停执行等待人工干预	通过配置或 interrupt 方法实现人机协作	⭐⭐⭐⭐
Time Travel	时间旅行，回溯和修改历史执行状态	get_state_history() 和 update_state() 方法实现	⭐⭐⭐⭐
Cycles	循环，图中节点可以回到之前的节点形成循环	通过边的定义实现，支持 Agent 的迭代执行	⭐⭐⭐⭐⭐
Controllability	可控性，开发者对执行流程的精确控制能力	通过显式定义节点、边和路由函数实现	⭐⭐⭐⭐⭐

📚 本章核心知识回顾

在开始问答之前,让我们快速回顾 Module 1 的知识地图:

Module 1: LangGraph 框架基础
├─ 1.1 框架对比 → 理解"为什么选择 LangGraph"
├─ 1.2 上手案例 → 建立"图思维"
├─ 1.3 LangChain 回顾 → 掌握基础组件
└─ 1.4 基础入门 → 系统化理解

核心概念:
  • Graph (图) → 执行引擎
  • State (状态) → 数据容器
  • Nodes (节点) → 状态转换函数
  • Edges (边) → 流程控制

🎯 复习问答 (15 题)

第一部分: 框架理解与选择 (5 题)

问题 1: LangGraph 的核心设计理念是什么?它解决了传统 Agent 框架的哪些问题?

💡 点击查看答案

答案:

LangGraph 的核心设计理念是图状态机 (Graph State Machine),将 Agent 的执行过程建模为有向图。

解决的核心问题:

可控性不足
- 传统框架: Agent 像"黑盒",执行流程难以预测
- LangGraph: 显式定义每个节点和边,流程完全可控
难以实现复杂逻辑
- 传统框架: 线性执行链,难以处理循环和复杂分支
- LangGraph: 支持任意图结构,包括循环
生产环境不可靠
- 传统框架: 缺乏断点、回溯、状态管理
- LangGraph: 提供 Breakpoints、Time Travel、持久化
调试困难
- 传统框架: 无法逐步执行,难以定位问题
- LangGraph: 可视化图结构,逐节点调试

关键洞察:

LangGraph = 可控性 + 灵活性 + 生产级特性

代码体现:

python

# 传统链式调用 (不可控)
result = chain.invoke(input)  # 黑盒执行

# LangGraph (完全可控)
graph.add_conditional_edges(
    "decision_node",
    router,  # 你定义路由逻辑
    {"path_a": "node_a", "path_b": "node_b"}
)

问题 2: 在以下场景中,应该选择哪个框架?请说明理由。

场景 A: 快速构建一个内容创作团队,包括研究员、作家、编辑三个角色 场景 B: 构建一个需要多次迭代的代码生成 Agent,要求可以自动修复错误 场景 C: 构建一个生产级的客服系统,需要精确控制流程和人工审批 场景 D: 快速实验一个简单的 Agent 交互模式

💡 点击查看答案

答案:

场景 A: CrewAI ✅

理由: 角色分工明确,任务线性流转,CrewAI 的 Agent + Task + Crew 模式最适合

代码示例:

python

researcher = Agent(role="研究员", goal="收集信息")
writer = Agent(role="作家", goal="撰写文章")
crew = Crew(agents=[researcher, writer])

场景 B: AutoGen ✅

理由: 代码生成和自动修复是 AutoGen 的强项,支持代码执行和自动迭代
特点: UserProxy 可以执行代码,Assistant 可以根据执行结果自动修复

场景 C: LangGraph ✅

理由:
- 生产级要求 → 需要可靠性和可观测性
- 精确控制流程 → 图状态机精确建模
- 人工审批 → Breakpoints 机制
关键特性: interrupt_before=["approval_node"]

场景 D: OpenAI Swarm ✅

理由: 实验性项目,最简单的实现,快速验证想法
注意: 不适合生产环境

决策原则:

快速原型 → CrewAI / Swarm
代码生成 → AutoGen
生产系统 → LangGraph

问题 3: LangGraph 相比 CrewAI 的最大优势是什么?什么情况下这个优势至关重要?

💡 点击查看答案

答案:

最大优势: 精确的流程控制和状态管理

具体体现:

条件分支

CrewAI: 线性任务流,分支能力有限
LangGraph: 任意复杂的条件路由

python

graph.add_conditional_edges(
    "analyzer",
    lambda state: "path_a" if state["score"] > 0.8 else "path_b",
    {"path_a": "high_score_handler", "path_b": "low_score_handler"}
)

循环执行

CrewAI: 不支持循环
LangGraph: 天然支持循环(如 Agent 重试)

python

# 可以添加从 "tools" 回到 "assistant" 的边
graph.add_edge("tools", "assistant")  # 形成循环

状态累积

CrewAI: 主要依赖消息传递
LangGraph: 完整的状态管理,支持任意复杂的状态

python

class AgentState(TypedDict):
    messages: list
    retry_count: int
    error_log: list
    context: dict

何时至关重要:

✅ 金融交易系统

需要多重审批
异常情况需要回退
每步决策都要记录

✅ 医疗诊断 Agent

诊断流程可能需要多次迭代
每个决策节点需要可审查
不同病情走不同路径

✅ 自动化运维系统

复杂的决策树
失败需要回滚
需要精确的状态追踪

关键洞察:

CrewAI 像"流水线",LangGraph 像"可编程的执行引擎"。当你需要的不仅是"协作",还有"精确控制"时,LangGraph 是唯一选择。

问题 4: 什么是"图思维"?它与传统的"链式思维"有什么本质区别?

💡 点击查看答案

答案:

图思维 vs 链式思维:

维度	链式思维	图思维
执行模式	线性: A→B→C→D	网络: 根据状态动态路由
决策点	预定义的固定流程	运行时动态决策
循环	不支持或难以实现	天然支持
状态	隐式,难以追踪	显式,完全可见
错误处理	中断整个链	可以跳转到错误处理节点

示例对比:

链式思维 (传统 Chain):

python

# 固定流程,无法根据结果改变路径
chain = prompt | llm | output_parser | tool_executor
result = chain.invoke(input)

图思维 (LangGraph):

python

# 根据 LLM 输出动态决定下一步
def router(state):
    last_message = state["messages"][-1]
    if has_tool_calls(last_message):
        return "tools"  # 需要调用工具
    elif needs_more_info(last_message):
        return "clarify"  # 需要澄清
    else:
        return END  # 直接结束

graph.add_conditional_edges("llm", router, {
    "tools": "tool_node",
    "clarify": "clarification_node"
})

图思维的本质:

状态驱动:

当前状态 + 节点处理 → 新状态 + 路由决策

非线性流程:

START → A → [决策] → B 或 C → [决策] → D 或回到 A → END

显式建模:

python

# 每个决策点都是显式的函数
def decide_next_step(state):
    # 你完全控制路由逻辑
    if state["confidence"] < 0.7:
        return "retry"
    return "continue"

实际应用价值:

场景: 智能客服

链式思维:
  用户输入 → LLM → 输出 (无法处理复杂情况)

图思维:
  用户输入 → 意图分类 → [路由]
                            ├→ FAQ: 直接回答
                            ├→ 技术问题: 调用知识库
                            ├→ 投诉: 转人工
                            └→ 其他: 澄清意图 (循环回到分类)

关键洞察:

图思维不仅是工具的改变,更是认知模式的升级。它让我们从"希望 AI 按预期工作"到"设计 AI 按预期工作"。

问题 5: LangGraph 如何实现 Human-in-the-Loop?举例说明其应用场景。

💡 点击查看答案

答案:

实现机制: Breakpoints (断点)

LangGraph 提供三种 Human-in-the-Loop 方式:

1. interrupt_before (执行前中断)

python

# 在执行 tools 节点前暂停,等待人工批准
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["tools"]  # 在工具调用前暂停
)

# 执行
config = {"configurable": {"thread_id": "1"}}
result = app.invoke(input, config)  # 暂停在工具调用前

# 人工审查后继续
result = app.invoke(None, config)  # 继续执行

2. interrupt_after (执行后中断)

python

# 在执行 tools 节点后暂停,查看结果
app = graph.compile(
    checkpointer=memory,
    interrupt_after=["tools"]  # 工具调用后暂停
)

3. 动态中断 (NodeInterrupt)

python

from langgraph.types import interrupt

def approval_node(state):
    # 根据条件决定是否需要人工审批
    if state["amount"] > 10000:
        decision = interrupt({
            "message": f"需要批准: ${state['amount']} 的支付",
            "data": state["payment_details"]
        })
        if decision != "approved":
            return {"status": "rejected"}
    return {"status": "approved"}

应用场景:

场景 1: 金融交易审批 💰

python

def payment_agent():
    # 构建图
    graph = StateGraph(PaymentState)
    graph.add_node("analyze", analyze_transaction)
    graph.add_node("execute", execute_payment)

    # 高风险交易需要人工审批
    def risk_router(state):
        if state["risk_score"] > 0.7:
            return "approval"  # 跳转到审批节点
        return "execute"  # 直接执行

    graph.add_conditional_edges("analyze", risk_router)

    # 在审批节点前中断
    app = graph.compile(interrupt_before=["approval"])

场景 2: 内容审核 📝

python

# 生成内容后,发布前需要人工审查
app = graph.compile(
    interrupt_after=["content_generation"],
    interrupt_before=["publish"]
)

# 工作流
# 1. 生成内容 (自动)
# 2. 暂停 → 人工审查
# 3. 批准后 → 发布

场景 3: 医疗诊断 🏥

python

class DiagnosisState(TypedDict):
    symptoms: list
    diagnosis: str
    confidence: float

def diagnosis_node(state):
    diagnosis, confidence = ai_diagnose(state["symptoms"])

    # 低置信度需要医生确认
    if confidence < 0.9:
        doctor_input = interrupt({
            "ai_diagnosis": diagnosis,
            "confidence": confidence,
            "request": "请医生确认或修正诊断"
        })
        diagnosis = doctor_input["final_diagnosis"]

    return {"diagnosis": diagnosis}

场景 4: 代码部署 🚀

python

# CI/CD 流程
graph.add_node("test", run_tests)
graph.add_node("deploy", deploy_to_production)

# 部署到生产前需要手动批准
app = graph.compile(interrupt_before=["deploy"])

# 工作流:
# 1. 运行测试 (自动)
# 2. 测试通过 → 暂停
# 3. DevOps 审查 → 批准
# 4. 部署到生产 (自动)

关键优势:

灵活性: 可以在任意节点中断
可恢复: 批准后从中断点继续,保持状态
可追溯: 所有人工决策都被记录
细粒度控制: 可以编辑状态后再继续

实现细节:

python

# 完整示例
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(
    checkpointer=memory,  # 必需: 保存状态
    interrupt_before=["critical_node"]
)

config = {"configurable": {"thread_id": "user123"}}

# 第一次调用: 执行到断点
result = app.invoke(input, config)

# 检查状态
state = app.get_state(config)
print(state.values)  # 当前状态
print(state.next)    # 下一个要执行的节点

# 人工决策后继续
result = app.invoke(None, config)  # None 表示继续执行

关键洞察:

Human-in-the-Loop 不是"功能",而是生产级 AI 系统的必需品。它让 AI 从"替代人"变为"辅助人",既提高效率,又保证安全。

第二部分: 核心概念 (5 题)

问题 6: 解释 LangGraph 中 State、Node、Edge 的关系。用一个类比说明它们如何协同工作。

💡 点击查看答案

答案:

核心关系:

State (状态)  = 共享的数据容器
Node (节点)   = 状态转换函数
Edge (边)     = 流程控制规则

数学表达:
  Node: State_old → State_new
  Edge: 决定下一个要执行的 Node
  Graph: 所有 Nodes 和 Edges 的组合

类比 1: 流水线工厂 🏭

State (状态)    = 产品及其加工进度
Node (节点)     = 加工站 (每个站执行特定操作)
Edge (边)       = 传送带路径

工作流程:
1. 原材料 (初始 State) 进入流水线
2. 第一个加工站 (Node) 处理 → 产品状态改变
3. 传送带 (Edge) 决定送到哪个加工站
4. 重复直到产品完成 (到达 END)

类比 2: RPG 游戏 🎮

State (状态)    = 角色属性 (HP, MP, 装备, 位置...)
Node (节点)     = 游戏事件 (战斗, 对话, 商店...)
Edge (边)       = 触发条件 (HP<50 → 逃跑, HP>50 → 继续战斗)

游戏流程:
1. 玩家状态: HP=100, 位置=森林
2. 遇敌事件 (Node) → 战斗 → HP=60
3. 检查 HP (Edge) → HP>50 → 继续探索 (Next Node)
4. 如果 HP<20 → 自动回城 (Different Node)

代码体现:

python

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. State: 数据容器
class GameState(TypedDict):
    player_hp: int
    enemy_hp: int
    location: str

# 2. Node: 状态转换函数
def battle_node(state: GameState) -> dict:
    # 读取状态
    player_hp = state["player_hp"]
    enemy_hp = state["enemy_hp"]

    # 处理逻辑 (战斗)
    player_hp -= 20  # 玩家受伤
    enemy_hp -= 30   # 敌人受伤

    # 返回更新 (部分更新!)
    return {
        "player_hp": player_hp,
        "enemy_hp": enemy_hp
    }

def heal_node(state: GameState) -> dict:
    return {"player_hp": min(state["player_hp"] + 50, 100)}

# 3. Edge: 流程控制
def decide_next_action(state: GameState) -> str:
    if state["player_hp"] < 30:
        return "heal"  # HP 低 → 治疗
    elif state["enemy_hp"] <= 0:
        return "end"   # 敌人死亡 → 结束
    else:
        return "battle"  # 继续战斗

# 4. 构建 Graph
graph = StateGraph(GameState)
graph.add_node("battle", battle_node)
graph.add_node("heal", heal_node)

graph.add_edge(START, "battle")
graph.add_conditional_edges(
    "battle",
    decide_next_action,
    {
        "battle": "battle",  # 循环
        "heal": "heal",
        "end": END
    }
)
graph.add_edge("heal", "battle")  # 治疗后继续战斗

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

协同工作流程:

1. 初始状态进入图
   State = {player_hp: 100, enemy_hp: 100}

2. START edge → battle node
   battle_node 执行 → 更新 State
   State = {player_hp: 80, enemy_hp: 70}

3. Conditional edge 检查状态
   decide_next_action(State) → "battle" (HP 还够)

4. battle edge → battle node (循环)
   battle_node 执行 → 更新 State
   State = {player_hp: 60, enemy_hp: 40}

5. 继续循环...直到某个条件满足

6. 最终 edge → END

关键洞察:

State 是"全局共享内存"
- 所有 Node 都能读取
- 每个 Node 只返回需要更新的部分
- LangGraph 自动合并更新

Node 是"纯函数"

python

# 理想的 Node 函数
def my_node(state: State) -> dict:
    # 1. 读取状态
    # 2. 执行逻辑
    # 3. 返回更新
    return {"field": new_value}

Edge 是"智能路由器"
- Normal Edge: 固定路径
- Conditional Edge: 根据状态动态选择路径

最佳实践:

python

# ✅ 好的设计
class State(TypedDict):
    # 清晰的字段定义
    messages: list[BaseMessage]
    user_info: dict
    next_action: str

def node(state: State) -> dict:
    # 只更新需要的字段
    return {"next_action": "call_tool"}

# ❌ 避免的模式
def bad_node(state: State) -> State:
    # 不要返回完整的 State
    state["field"] = "value"  # 不要直接修改
    return state  # 不推荐

问题 7: 什么是 Channels?它在状态管理中扮演什么角色?

💡 点击查看答案

答案:

Channels (通道) 是 LangGraph 状态管理的核心机制。

定义:

State 中的每个键 (key) 就是一个 Channel
Channel 是独立的数据通道,有自己的更新规则

示例:

python

class State(TypedDict):
    name: str          # name 通道
    messages: list     # messages 通道
    count: int         # count 通道

这个 State 有 3 个 Channels: name, messages, count

Channel 的特性:

1. 独立更新

python

# Node 1 只更新 name
def node_1(state):
    return {"name": "Alice"}  # 其他通道不变

# Node 2 只更新 messages
def node_2(state):
    return {"messages": [new_message]}  # 其他通道不变

2. 默认覆盖行为

python

state = {"name": "Alice", "count": 1}

# Node 返回
return {"name": "Bob", "count": 2}

# 结果: 完全覆盖
state = {"name": "Bob", "count": 2}

3. Reducer 函数 (关键!)

对于列表类型,通常需要追加而不是覆盖:

python

from typing import Annotated
from operator import add

class State(TypedDict):
    messages: Annotated[list, add]  # 使用 add reducer
    #         ^^^^^^^^^^^^^^^^^^^^
    #         这个通道使用 add 函数合并更新

# 现在的行为:
state = {"messages": [msg1, msg2]}

# Node 返回
return {"messages": [msg3]}

# 结果: 追加 (不是覆盖!)
state = {"messages": [msg1, msg2, msg3]}

常用 Reducers:

python

from operator import add
from typing import Annotated

# 1. 列表追加
messages: Annotated[list, add]

# 2. 字典合并
context: Annotated[dict, lambda x, y: {**x, **y}]

# 3. 自定义 reducer
def custom_reducer(current, update):
    # current: 当前值
    # update: 节点返回的值
    return current + update  # 你的合并逻辑

field: Annotated[type, custom_reducer]

实际应用:

场景: 多轮对话 Agent

python

from langchain_core.messages import BaseMessage
from typing import Annotated
from operator import add

class ChatState(TypedDict):
    # messages 通道: 追加消息,不覆盖
    messages: Annotated[list[BaseMessage], add]

    # user_info 通道: 覆盖更新
    user_info: dict

    # turn_count 通道: 递增
    turn_count: Annotated[int, lambda curr, upd: curr + upd]

# Node 1: 添加用户消息
def user_input_node(state):
    return {
        "messages": [HumanMessage(content="Hello")],
        "turn_count": 1  # 递增 1
    }
    # messages 会追加,不会覆盖!
    # turn_count 会相加,不会覆盖!

# Node 2: 添加 AI 消息
def ai_response_node(state):
    return {
        "messages": [AIMessage(content="Hi there!")],
        "turn_count": 1
    }
    # 又追加了一条消息
    # turn_count 又加 1

# 执行后:
# state = {
#     "messages": [HumanMessage("Hello"), AIMessage("Hi there!")],
#     "user_info": {...},  # 未更新,保持原值
#     "turn_count": 2      # 1 + 1 = 2
# }

Channels 的高级用法:

1. 条件更新

python

def node(state):
    updates = {}

    if state["score"] > 0.8:
        updates["status"] = "passed"  # 只更新 status 通道

    if len(state["messages"]) > 10:
        updates["messages"] = state["messages"][-5:]  # 只保留最后 5 条

    return updates  # 可以返回任意通道的组合

2. 依赖关系

python

class State(TypedDict):
    raw_data: str
    processed_data: str  # 依赖 raw_data

def process_node(state):
    # 基于 raw_data 通道计算 processed_data 通道
    processed = transform(state["raw_data"])
    return {"processed_data": processed}

3. 通道隔离

python

# 不同 Node 可以关注不同的 Channels
def node_a(state):
    # 只关心 channel_a
    return {"channel_a": value_a}

def node_b(state):
    # 只关心 channel_b
    return {"channel_b": value_b}

# 它们不会互相干扰

可视化理解:

State (状态容器)
├─ Channel: name (覆盖模式)
│   current: "Alice"
│   update:  "Bob"
│   result:  "Bob" ← 直接覆盖
│
├─ Channel: messages (追加模式)
│   current: [msg1, msg2]
│   update:  [msg3]
│   result:  [msg1, msg2, msg3] ← 使用 add reducer
│
└─ Channel: count (累加模式)
    current: 5
    update:  3
    result:  8 ← 使用 lambda curr, upd: curr + upd

最佳实践:

python

# ✅ 明确指定 Reducer
from typing import Annotated
from operator import add

class State(TypedDict):
    # 列表: 用 add
    messages: Annotated[list, add]

    # 字典: 用合并函数
    context: Annotated[dict, lambda x, y: {**x, **y}]

    # 数字: 用累加或覆盖 (根据需求)
    count: Annotated[int, lambda curr, upd: curr + upd]  # 累加
    total: int  # 覆盖 (默认行为)

# ❌ 忘记 Reducer 导致的问题
class BadState(TypedDict):
    messages: list  # 没有 add → 会被覆盖!

def node(state):
    return {"messages": [new_msg]}
    # 结果: state["messages"] = [new_msg]  ← 旧消息丢失!

关键洞察:

Channels 实现模块化状态管理
- 每个 Channel 独立更新
- 避免了"全局状态污染"
Reducer 决定合并策略
- 默认: 覆盖
- 列表: 通常用 add
- 自定义: 任意逻辑

设计原则: 最小更新

python

# 只返回改变的 Channels
return {"field_a": new_value}
# 而不是返回整个 State

问题 8: 写出一个完整的 LangGraph 应用的最小代码模板,并解释每个部分的作用。

💡 点击查看答案

答案:

完整的最小模板:

python

# ========== 1. 导入必需的包 ==========
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# ========== 2. 定义状态 Schema ==========
class State(TypedDict):
    """
    状态定义: 图中流动的数据结构
    - 所有节点共享这个状态
    - 节点通过返回字典来更新状态
    """
    input: str    # 输入数据
    output: str   # 输出数据

# ========== 3. 定义节点函数 ==========
def processing_node(state: State) -> dict:
    """
    节点函数: State → dict
    - 接收完整的 State
    - 返回要更新的字段 (部分更新)
    """
    # 读取状态
    user_input = state["input"]

    # 执行逻辑
    processed = f"Processed: {user_input}"

    # 返回更新 (只返回改变的字段)
    return {"output": processed}

# ========== 4. 构建图 ==========
# 4.1 创建图实例
graph = StateGraph(State)

# 4.2 添加节点
graph.add_node("process", processing_node)

# 4.3 添加边 (定义流程)
graph.add_edge(START, "process")  # START → process
graph.add_edge("process", END)    # process → END

# ========== 5. 编译图 ==========
app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# ========== 6. 运行图 ==========
result = app.invoke({"input": "Hello, World!"})

# ========== 7. 查看结果 ==========
print(result)
# 输出: {'input': 'Hello, World!', 'output': 'Processed: Hello, World!'}

各部分详细解释:

1. 导入 📦

python

from typing_extensions import TypedDict  # 定义类型化字典
from langgraph.graph import StateGraph, START, END  # 核心类和常量

TypedDict: Python 类型提示,定义状态结构
StateGraph: 图构建器
START/END: 特殊节点标记

2. 状态定义 📝

python

class State(TypedDict):
    input: str
    output: str

作用:

定义图中流动的数据结构
类型提示帮助 IDE 自动补全和类型检查
所有节点共享这个状态

类比: State 是"共享内存",所有节点都能读写

3. 节点函数 ⚙️

python

def processing_node(state: State) -> dict:
    user_input = state["input"]         # 读取
    processed = f"Processed: {user_input}"  # 处理
    return {"output": processed}        # 更新

作用:

执行具体的业务逻辑
签名: (State) -> dict
返回: 只返回需要更新的字段

关键点:

✅ 返回 dict,不是 State
✅ 部分更新,不是全量替换
✅ 纯函数,无副作用

4. 构建图 🔧

python

graph = StateGraph(State)              # 创建图
graph.add_node("process", processing_node)  # 添加节点
graph.add_edge(START, "process")       # 定义流程
graph.add_edge("process", END)

作用:

StateGraph(State): 创建图构建器,指定状态类型
add_node: 注册节点函数
add_edge: 定义执行顺序

执行流程:

START → "process" → END

5. 编译 🔨

python

app = graph.compile()

作用:

将图定义转换为可执行的应用
验证图的正确性 (无悬空节点、循环检测等)
返回 CompiledGraph 对象

类比: 编译源代码为可执行文件

6. 运行 ▶️

python

result = app.invoke({"input": "Hello, World!"})

作用:

执行图,传入初始状态
返回最终状态

执行过程:

1. 初始状态: {"input": "Hello, World!"}
2. START → process 节点
3. process_node 执行 → {"output": "Processed: ..."}
4. 状态合并: {"input": "...", "output": "..."}
5. process → END
6. 返回最终状态

带条件路由的完整模板:

python

from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

# 状态定义
class State(TypedDict):
    input: str
    score: float
    output: str

# 节点函数
def analyze_node(state: State) -> dict:
    """分析输入,打分"""
    text = state["input"]
    score = len(text) / 100.0  # 简单打分逻辑
    return {"score": score}

def high_score_node(state: State) -> dict:
    return {"output": f"High quality: {state['input']}"}

def low_score_node(state: State) -> dict:
    return {"output": f"Needs improvement: {state['input']}"}

# 路由函数
def router(state: State) -> Literal["high", "low"]:
    """根据分数路由"""
    return "high" if state["score"] > 0.5 else "low"

# 构建图
graph = StateGraph(State)

# 添加节点
graph.add_node("analyze", analyze_node)
graph.add_node("high_quality", high_score_node)
graph.add_node("low_quality", low_score_node)

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

# 条件边
graph.add_conditional_edges(
    "analyze",           # 源节点
    router,              # 路由函数
    {                    # 路径映射
        "high": "high_quality",
        "low": "low_quality"
    }
)

graph.add_edge("high_quality", END)
graph.add_edge("low_quality", END)

# 编译和运行
app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({"input": "This is a test message"})
print(result)

关键洞察:

模板化思维

定义 State → 定义 Nodes → 连接 Edges → 编译 → 运行

状态是核心
- 所有逻辑围绕状态展开
- 节点通过状态通信
声明式构建
- 先声明图结构
- 后执行具体逻辑

问题 9: 条件边 (Conditional Edge) 的路由函数有什么要求?如何处理多分支路由?

💡 点击查看答案

答案:

路由函数的要求:

1. 函数签名

python

from typing import Literal

def router(state: State) -> Literal["path_a", "path_b", "path_c"]:
    #          ^^^^^ 输入: State
    #                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^ 输出: 路径名
    pass

要求:

✅ 输入: 接收 State
✅ 输出: 返回字符串,表示下一个节点的名称
✅ 类型提示: 使用 Literal 明确所有可能的路径 (推荐但非必需)

2. 返回值

python

# ✅ 正确: 返回节点名
return "node_a"

# ✅ 正确: 返回 END
return END

# ❌ 错误: 返回其他类型
return True  # 不行!
return 123   # 不行!

3. 纯函数

python

# ✅ 好的路由函数: 纯函数,无副作用
def good_router(state):
    if state["score"] > 0.8:
        return "high"
    return "low"

# ❌ 避免: 有副作用
def bad_router(state):
    print("Routing...")  # 副作用: 打印
    state["routed"] = True  # 副作用: 修改状态
    return "next"

多分支路由的实现:

方式 1: if-elif-else (基础)

python

def multi_branch_router(state: State) -> Literal["path_a", "path_b", "path_c", "end"]:
    score = state["score"]

    if score > 0.9:
        return "path_a"  # 优秀
    elif score > 0.7:
        return "path_b"  # 良好
    elif score > 0.5:
        return "path_c"  # 及格
    else:
        return "end"     # 不及格,直接结束

# 使用
graph.add_conditional_edges(
    "classifier",
    multi_branch_router,
    {
        "path_a": "excellent_handler",
        "path_b": "good_handler",
        "path_c": "pass_handler",
        "end": END
    }
)

方式 2: 字典映射 (优雅)

python

def intent_router(state: State) -> str:
    intent = state["intent"]

    route_map = {
        "weather": "weather_tool",
        "booking": "booking_tool",
        "faq": "faq_handler",
        "complaint": "human_agent",
    }

    # 提供默认路径
    return route_map.get(intent, "fallback")

# 使用
graph.add_conditional_edges(
    "intent_classifier",
    intent_router,
    {
        "weather_tool": "weather_node",
        "booking_tool": "booking_node",
        "faq_handler": "faq_node",
        "human_agent": "handoff_node",
        "fallback": "general_response_node"
    }
)

方式 3: 基于模式匹配 (Python 3.10+)

python

def pattern_router(state: State) -> str:
    match state["category"]:
        case "urgent":
            return "priority_handler"
        case "normal":
            return "standard_handler"
        case "low":
            return "queue_handler"
        case _:
            return "default_handler"

方式 4: 复杂逻辑 (组合条件)

python

def complex_router(state: State) -> str:
    user_type = state["user_type"]
    request_type = state["request_type"]
    priority = state["priority"]

    # VIP 用户
    if user_type == "vip":
        return "vip_handler"

    # 紧急请求
    if priority == "urgent":
        if request_type == "technical":
            return "tech_urgent"
        else:
            return "general_urgent"

    # 普通请求
    if request_type == "technical":
        return "tech_support"
    elif request_type == "billing":
        return "billing_support"
    else:
        return "general_support"

实际案例: 智能客服系统

python

from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class CustomerServiceState(TypedDict):
    user_message: str
    intent: str
    sentiment: str  # positive/negative/neutral
    user_tier: str  # vip/premium/regular
    requires_human: bool

def classify_node(state):
    """意图分类和情感分析"""
    message = state["user_message"]

    # 简化的分类逻辑
    intent = "complaint" if "problem" in message.lower() else "inquiry"
    sentiment = "negative" if "bad" in message.lower() else "neutral"

    return {
        "intent": intent,
        "sentiment": sentiment
    }

def route_request(state) -> Literal["auto", "human", "vip", "end"]:
    """智能路由逻辑"""

    # VIP 用户直接转人工
    if state["user_tier"] == "vip":
        return "vip"

    # 投诉 + 负面情绪 → 人工
    if state["intent"] == "complaint" and state["sentiment"] == "negative":
        return "human"

    # 标记需要人工的情况
    if state.get("requires_human", False):
        return "human"

    # 一般询问 → 自动回复
    if state["intent"] == "inquiry":
        return "auto"

    # 其他情况结束
    return "end"

# 构建图
graph = StateGraph(CustomerServiceState)
graph.add_node("classify", classify_node)
graph.add_node("auto_response", lambda s: {"output": "自动回复"})
graph.add_node("human_handoff", lambda s: {"output": "转人工"})
graph.add_node("vip_service", lambda s: {"output": "VIP 服务"})

graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_request,
    {
        "auto": "auto_response",
        "human": "human_handoff",
        "vip": "vip_service",
        "end": END
    }
)
graph.add_edge("auto_response", END)
graph.add_edge("human_handoff", END)
graph.add_edge("vip_service", END)

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 测试
result = app.invoke({
    "user_message": "I have a problem with my order",
    "user_tier": "regular"
})

最佳实践:

1. 使用 Literal 类型提示

python

# ✅ 推荐: 明确所有路径
def router(state) -> Literal["path_a", "path_b", "path_c"]:
    ...

# ⚠️ 可行但不推荐: 没有类型提示
def router(state) -> str:
    ...

2. 提供默认路径

python

def safe_router(state):
    route_map = {...}
    return route_map.get(state["intent"], "fallback")  # 默认路径

3. 记录路由决策 (调试)

python

def logged_router(state):
    decision = make_routing_decision(state)
    print(f"Routing to: {decision}")  # 生产环境用 logging
    return decision

4. 验证路径存在

python

# 在图中定义所有路径
graph.add_conditional_edges(
    "source",
    router,
    {
        "path_a": "node_a",
        "path_b": "node_b",
        # 确保 router 返回的所有值都在这里定义
    }
)

关键洞察:

路由函数是"流程控制的大脑"
- 决定 Agent 的下一步行动
- 是 LangGraph 动态性的核心
状态驱动路由
```
State → Router → Next Node
```
1

可测试性

python

# 路由函数是纯函数,容易测试
assert router({"score": 0.9}) == "high"
assert router({"score": 0.3}) == "low"

问题 10: 在 LangGraph 中,如何实现循环?举例说明一个需要循环的真实场景。

💡 点击查看答案

答案:

实现循环的方法:

LangGraph 中的循环通过边连接实现:

python

# 循环的本质: 从后面的节点连回前面的节点
graph.add_edge("node_b", "node_a")  # node_b → node_a (形成循环)

基础循环结构:

python

from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: int
    max_iterations: int

def increment_node(state):
    return {"count": state["count"] + 1}

def check_node(state) -> Literal["continue", "end"]:
    if state["count"] < state["max_iterations"]:
        return "continue"  # 继续循环
    return "end"           # 跳出循环

graph = StateGraph(State)
graph.add_node("increment", increment_node)

graph.add_edge(START, "increment")
graph.add_conditional_edges(
    "increment",
    check_node,
    {
        "continue": "increment",  # 回到自己,形成循环
        "end": END
    }
)

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

result = app.invoke({"count": 0, "max_iterations": 5})
# 执行流程: START → increment → increment → ... → END
# count: 0 → 1 → 2 → 3 → 4 → 5

真实场景 1: ReAct Agent (Reasoning + Acting)

这是最经典的循环场景:

python

from langchain_core.messages import BaseMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from typing import Annotated
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add]

# LLM with tools
llm = ChatOpenAI(model="gpt-4")
tools = [search_tool, calculator_tool]
llm_with_tools = llm.bind_tools(tools)

def llm_node(state):
    """LLM 决策: 调用工具或给出最终答案"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state):
    """执行工具调用"""
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls

    # 执行所有工具调用
    results = []
    for tool_call in tool_calls:
        tool = get_tool(tool_call["name"])
        result = tool.invoke(tool_call["args"])
        results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": results}

def should_continue(state) -> Literal["tools", "end"]:
    """决定是继续调用工具,还是结束"""
    last_message = state["messages"][-1]

    # 如果 LLM 决定调用工具,继续循环
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # 否则结束
    return "end"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "llm")
graph.add_conditional_edges(
    "llm",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)
graph.add_edge("tools", "llm")  # 🔁 关键: 形成循环

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 运行
result = app.invoke({
    "messages": [HumanMessage(content="What's 25 * 4, and search for the weather?")]
})

# 执行流程:
# 1. llm → 决定调用 calculator 和 search
# 2. tools → 执行两个工具
# 3. llm → 看到工具结果,决定调用更多工具或给出答案
# 4. ... (可能多次循环)
# 5. llm → 给出最终答案 → END

执行流程可视化:

START
  ↓
[llm] "我需要计算 25*4 和查询天气"
  ↓ (should_continue → "tools")
[tools] 执行 calculator(25, 4) 和 search("weather")
  ↓
[llm] "结果是 100,天气是晴天。让我总结答案"
  ↓ (should_continue → "end")
END

真实场景 2: 错误重试机制

python

class RetryState(TypedDict):
    task: str
    result: Optional[str]
    error: Optional[str]
    retry_count: int
    max_retries: int

def execute_task_node(state):
    """执行任务 (可能失败)"""
    try:
        result = risky_operation(state["task"])
        return {"result": result, "error": None}
    except Exception as e:
        return {
            "error": str(e),
            "retry_count": state["retry_count"] + 1
        }

def check_retry(state) -> Literal["retry", "failed", "success"]:
    """检查是否需要重试"""
    # 成功
    if state["result"] is not None:
        return "success"

    # 达到最大重试次数
    if state["retry_count"] >= state["max_retries"]:
        return "failed"

    # 继续重试
    return "retry"

graph = StateGraph(RetryState)
graph.add_node("execute", execute_task_node)
graph.add_node("success_handler", lambda s: {"result": "Task completed"})
graph.add_node("failure_handler", lambda s: {"result": "Task failed"})

graph.add_edge(START, "execute")
graph.add_conditional_edges(
    "execute",
    check_retry,
    {
        "retry": "execute",       # 🔁 重试: 回到 execute
        "success": "success_handler",
        "failed": "failure_handler"
    }
)
graph.add_edge("success_handler", END)
graph.add_edge("failure_handler", END)

真实场景 3: 对话澄清循环

python

class ClarificationState(TypedDict):
    user_messages: list[str]
    bot_messages: list[str]
    intent_clear: bool
    clarification_attempts: int

def understand_intent_node(state):
    """尝试理解用户意图"""
    latest_message = state["user_messages"][-1]
    intent_clear = is_intent_clear(latest_message)
    return {"intent_clear": intent_clear}

def ask_clarification_node(state):
    """询问澄清问题"""
    question = generate_clarification_question(state)
    return {
        "bot_messages": [question],
        "clarification_attempts": state["clarification_attempts"] + 1
    }

def final_response_node(state):
    """给出最终答案"""
    response = generate_response(state)
    return {"bot_messages": [response]}

def should_clarify(state) -> Literal["clarify", "respond"]:
    """决定是否需要澄清"""
    if state["intent_clear"]:
        return "respond"  # 意图清楚,直接回答

    if state["clarification_attempts"] >= 2:
        return "respond"  # 尝试太多次,直接回答

    return "clarify"  # 继续澄清

graph = StateGraph(ClarificationState)
graph.add_node("understand", understand_intent_node)
graph.add_node("clarify", ask_clarification_node)
graph.add_node("respond", final_response_node)

graph.add_edge(START, "understand")
graph.add_conditional_edges(
    "understand",
    should_clarify,
    {
        "clarify": "clarify",
        "respond": "respond"
    }
)
graph.add_edge("clarify", "understand")  # 🔁 澄清后重新理解
graph.add_edge("respond", END)

# 对话流程:
# User: "我想订个东西"
# Bot: "请问您想订什么?" (clarify)
# User: "机票"
# Bot: "好的,请问是哪个城市?" (clarify)
# User: "北京到上海"
# Bot: "为您查询北京到上海的机票..." (respond)

循环控制的最佳实践:

1. 防止无限循环

python

class State(TypedDict):
    iteration_count: int
    max_iterations: int  # 强制上限

def router(state):
    if state["iteration_count"] >= state["max_iterations"]:
        return "end"  # 强制退出
    if should_continue(state):
        return "loop"
    return "end"

2. 记录循环路径 (调试)

python

def debug_node(state):
    print(f"Iteration {state['iteration_count']}: {state['current_step']}")
    return {}

3. 渐进式退出条件

python

def should_continue(state):
    # 多个退出条件
    if state["found_answer"]:
        return "end"
    if state["retry_count"] > 3:
        return "end"
    if state["time_elapsed"] > 30:
        return "end"
    return "continue"

关键洞察:

循环是 Agent 智能的体现
- ReAct: 思考 → 行动 → 观察 → 思考 (循环)
- 自我修正: 尝试 → 检查 → 重试 (循环)
LangGraph 的循环是"可控循环"
- 不同于递归或 while 循环
- 每一步都是可观测的
- 可以随时中断和恢复

循环 + 条件边 = 强大的 Agent 模式

决策节点 ⇄ 执行节点
    ↓ (条件满足)
  结束

第三部分: LangChain 基础 (3 题)

问题 11: 解释 LangChain 中的 Messages 类型。为什么 LangGraph 中经常使用 `messages: list[BaseMessage]` 作为状态字段?

💡 点击查看答案

答案:

LangChain Messages 类型体系:

python

from langchain_core.messages import (
    BaseMessage,      # 基类
    HumanMessage,     # 用户消息
    AIMessage,        # AI 回复
    SystemMessage,    # 系统提示
    ToolMessage,      # 工具执行结果
    FunctionMessage   # 函数调用结果 (已废弃,用 ToolMessage)
)

每种消息的作用:

1. SystemMessage 🤖

python

SystemMessage(content="You are a helpful assistant")

用途: 定义 AI 的角色和行为
位置: 通常在对话开始
特点: 不显示给用户,只影响 AI 行为

2. HumanMessage 👤

python

HumanMessage(content="What's the weather today?")

用途: 用户的输入
来源: 用户界面
特点: 驱动对话的进展

3. AIMessage 🧠

python

AIMessage(content="Let me check the weather for you.")

# 带工具调用
AIMessage(
    content="",
    tool_calls=[{
        "id": "call_123",
        "name": "get_weather",
        "args": {"city": "Beijing"}
    }]
)

用途: AI 的回复
特殊: 可以包含 tool_calls (工具调用请求)

4. ToolMessage 🔧

python

ToolMessage(
    content='{"temp": 22, "condition": "sunny"}',
    tool_call_id="call_123"
)

用途: 工具执行的结果
关联: 通过 tool_call_id 关联到 AIMessage 的 tool_call

消息流示例:

python

messages = [
    SystemMessage(content="You are a helpful assistant"),
    HumanMessage(content="What's 25 * 4?"),
    AIMessage(
        content="",
        tool_calls=[{"id": "1", "name": "calculator", "args": {"a": 25, "b": 4}}]
    ),
    ToolMessage(content="100", tool_call_id="1"),
    AIMessage(content="The answer is 100")
]

对话流程:

System → 设定角色
Human → 提问
AI → 决定调用工具
Tool → 返回结果
AI → 基于结果回答

为什么 LangGraph 使用 messages: list[BaseMessage]?

原因 1: 统一的对话历史格式 📚

python

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add]  # 所有消息类型的列表

优势:

通用性: 支持所有类型的消息
扩展性: 新增消息类型无需改变状态结构
兼容性: 直接传给 LLM API

原因 2: 与 LLM API 无缝集成 🔗

python

def llm_node(state):
    # LangChain 的 LLM 直接接受 messages 列表
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

LLM API 的输入格式:

json

{
  "messages": [
    {"role": "system", "content": "..."},
    {"role": "user", "content": "..."},
    {"role": "assistant", "content": "..."}
  ]
}

LangChain 的 Messages 自动转换为这种格式!

原因 3: 支持 Tool Calling 工作流 🛠️

python

# 完整的 ReAct 循环
def agent_node(state):
    # 1. LLM 决策 (可能调用工具)
    response = llm.invoke(state["messages"])
    # response 可能包含 tool_calls
    return {"messages": [response]}

def tool_node(state):
    # 2. 执行工具
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        result = execute_tool(tool_call)
        results.append(ToolMessage(
            content=result,
            tool_call_id=tool_call["id"]
        ))
    return {"messages": results}

# 循环: agent → tool → agent → tool → ... → agent (最终答案)

Messages 列表的演变:

python

[
    HumanMessage("What's 25*4 and search weather?"),
]
↓ LLM 决策
[
    HumanMessage(...),
    AIMessage(tool_calls=[calculator, search])
]
↓ 工具执行
[
    HumanMessage(...),
    AIMessage(tool_calls=[...]),
    ToolMessage(content="100", tool_call_id="1"),
    ToolMessage(content="Sunny", tool_call_id="2")
]
↓ LLM 总结
[
    ...,
    AIMessage("The answer is 100, and the weather is sunny")
]

原因 4: Reducer 机制的完美应用 ➕

python

from operator import add

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    #         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 使用 add reducer

为什么用 add:

python

# 不用 add (默认覆盖)
state = {"messages": [msg1, msg2]}
node_return = {"messages": [msg3]}
# 结果: {"messages": [msg3]}  ← 旧消息丢失!

# 使用 add (追加)
state = {"messages": [msg1, msg2]}
node_return = {"messages": [msg3]}
# 结果: {"messages": [msg1, msg2, msg3]}  ← 保留历史!

实际应用模式:

模式 1: 简单对话

python

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add]

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

# 使用
result = app.invoke({
    "messages": [HumanMessage("Hello")]
})
# 结果: messages = [HumanMessage("Hello"), AIMessage("Hi there!")]

模式 2: 带系统提示

python

def add_system_message(state):
    if not state["messages"] or state["messages"][0].type != "system":
        return {"messages": [SystemMessage("You are a helpful assistant")]}
    return {}

graph.add_edge(START, "add_system")
graph.add_edge("add_system", "chatbot")

模式 3: 消息修剪 (避免超长)

python

def trim_messages_node(state):
    messages = state["messages"]
    if len(messages) > 10:
        # 保留系统消息 + 最后 9 条
        system_msg = [m for m in messages if m.type == "system"]
        recent_msgs = messages[-9:]
        return {"messages": system_msg + recent_msgs}
    return {}

最佳实践:

python

# ✅ 推荐: 使用 Annotated + add
from operator import add
from typing import Annotated

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add]

# ✅ 节点返回新消息
def node(state):
    new_msg = AIMessage(content="...")
    return {"messages": [new_msg]}  # 会自动追加

# ❌ 避免: 直接修改 messages
def bad_node(state):
    state["messages"].append(new_msg)  # 不推荐
    return {}

# ❌ 避免: 返回所有消息
def bad_node2(state):
    all_messages = state["messages"] + [new_msg]
    return {"messages": all_messages}  # 冗余

关键洞察:

Messages 是对话的"DNA"
- 完整记录对话历史
- 包含所有上下文信息
list[BaseMessage] 是"通用接口"
- 与 LLM API 无缝对接
- 支持所有对话模式
add Reducer 是"历史累积器"
- 自动维护对话历史
- 避免手动管理列表

问题 12: 什么是 Tools?如何在 LangGraph 中使用工具?解释 `bind_tools` 的作用。

💡 点击查看答案

答案:

Tools (工具) 的定义:

Tools 是 Agent 与外部世界交互的"能力",使 AI 不再局限于文本生成,可以执行实际操作。

工具的类型:

🔍 搜索工具: Google 搜索、Wikipedia 查询
🧮 计算工具: 数学计算、数据分析
📊 数据库工具: SQL 查询、数据提取
🌐 API 工具: 天气 API、支付 API
💻 系统工具: 文件操作、命令执行

定义工具的方法:

方法 1: 使用 @tool 装饰器 (推荐)

python

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers.

    Args:
        a: First number
        b: Second number
    """
    return a * b

@tool
def search(query: str) -> str:
    """Search for information.

    Args:
        query: The search query
    """
    return f"Search results for: {query}"

# tool 对象包含:
# - name: "multiply"
# - description: docstring 的第一行
# - args_schema: 自动从参数推断

重要: docstring 是工具的"说明书",LLM 根据它决定何时调用工具!

方法 2: 使用 Tool 类

python

from langchain_core.tools import Tool

def my_function(input: str) -> str:
    return f"Processed: {input}"

tool = Tool(
    name="my_tool",
    description="A tool that processes input",
    func=my_function
)

方法 3: 使用 StructuredTool (复杂参数)

python

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="The search query")
    num_results: int = Field(default=5, description="Number of results")

def search_function(query: str, num_results: int) -> str:
    return f"Found {num_results} results for '{query}'"

search_tool = StructuredTool(
    name="search",
    description="Search the web",
    func=search_function,
    args_schema=SearchInput
)

bind_tools 的作用:

bind_tools 将工具"绑定"到 LLM,使 LLM 知道有哪些工具可用。

不使用 bind_tools:

python

llm = ChatOpenAI(model="gpt-4")
response = llm.invoke("What's 25 * 4?")
# LLM 只能文本回答: "25 * 4 = 100"
# 不会调用计算器工具

使用 bind_tools:

python

llm = ChatOpenAI(model="gpt-4")
tools = [multiply, search]
llm_with_tools = llm.bind_tools(tools)  # 🔑 关键步骤

response = llm_with_tools.invoke("What's 25 * 4?")
# LLM 决定调用 multiply 工具
# response.tool_calls = [{
#     "id": "call_123",
#     "name": "multiply",
#     "args": {"a": 25, "b": 4}
# }]

bind_tools 做了什么:

转换工具定义 → OpenAI function calling 格式
发送给 LLM → LLM 知道有哪些工具
LLM 返回 → 决定是否调用工具

OpenAI API 格式:

json

{
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "multiply",
        "description": "Multiply two numbers",
        "parameters": {
          "type": "object",
          "properties": {
            "a": {"type": "integer"},
            "b": {"type": "integer"}
          }
        }
      }
    }
  ]
}

在 LangGraph 中使用工具:

完整的 ReAct Agent 实现:

python

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import Annotated
from operator import add

# 1. 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

@tool
def add_numbers(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

tools = [multiply, add_numbers]

# 2. 创建带工具的 LLM
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)

# 3. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add]

# 4. 定义节点
def agent_node(state):
    """LLM 决策节点"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 使用 LangGraph 内置的 ToolNode (自动执行工具)
tool_node = ToolNode(tools)

# 5. 路由函数
def should_continue(state) -> Literal["tools", "end"]:
    """检查是否需要调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

# 6. 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)
graph.add_edge("tools", "agent")  # 工具执行后回到 agent

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 7. 使用
result = app.invoke({
    "messages": [HumanMessage("What's 25 * 4 plus 10?")]
})

# 执行流程:
# User: "What's 25 * 4 plus 10?"
#   ↓
# Agent: "我需要先算 25*4, 然后加10"
#   → tool_calls: [multiply(25, 4)]
#   ↓
# Tools: 执行 multiply(25, 4) → 100
#   → ToolMessage(content="100", tool_call_id="...")
#   ↓
# Agent: "25*4=100, 现在需要 100+10"
#   → tool_calls: [add_numbers(100, 10)]
#   ↓
# Tools: 执行 add_numbers(100, 10) → 110
#   ↓
# Agent: "最终答案是 110"
#   → 不调用工具,直接返回
#   ↓
# END

手动实现 Tool Node (理解原理):

python

def manual_tool_node(state):
    """手动执行工具调用"""
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls

    # 工具映射
    tool_map = {
        "multiply": multiply,
        "add_numbers": add_numbers
    }

    # 执行所有工具调用
    results = []
    for tool_call in tool_calls:
        # 1. 找到工具函数
        tool_func = tool_map[tool_call["name"]]

        # 2. 执行工具
        result = tool_func.invoke(tool_call["args"])

        # 3. 构建 ToolMessage
        results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))

    return {"messages": results}

工具设计的最佳实践:

1. 清晰的 docstring

python

@tool
def good_tool(query: str, max_results: int = 5) -> str:
    """Search for information on the web.

    Use this tool when you need to find current information,
    news, or facts that you don't have in your knowledge base.

    Args:
        query: The search query (be specific!)
        max_results: Number of results to return (1-10)

    Returns:
        A formatted list of search results
    """
    pass

# ❌ 不好的 docstring
@tool
def bad_tool(query: str) -> str:
    """Search."""  # 太简单,LLM 不知道何时用
    pass

2. 类型提示

python

@tool
def typed_tool(a: int, b: int) -> int:
    #             ^^^       ^^^
    # 类型提示帮助 LLM 理解参数类型
    """Add two integers."""
    return a + b

3. 错误处理

python

@tool
def safe_tool(url: str) -> str:
    """Fetch content from a URL."""
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        return f"Error fetching URL: {str(e)}"

4. 工具返回格式

python

# ✅ 返回字符串 (LLM 容易理解)
@tool
def format_result(data: dict) -> str:
    """Process data and return formatted string."""
    return json.dumps(data, indent=2)

# ⚠️ 返回复杂对象 (可能需要额外处理)
@tool
def raw_result(data: dict) -> dict:
    return data  # LLM 会收到 str(data)

关键洞察:

Tools 让 AI 从"语言模型"变为"Agent"

LLM → 只能文本生成
LLM + Tools → 可以执行实际任务

bind_tools 是"能力注册"

bind_tools(tools) → 告诉 LLM 它有哪些"超能力"

LangGraph 的工具循环是"思考-行动循环"

思考 (Agent Node) → 决定调用工具
行动 (Tool Node) → 执行工具
观察 (回到 Agent Node) → 看结果,决定下一步

ToolNode 是便捷封装

python

# 不用手动写工具执行逻辑
tool_node = ToolNode(tools)  # 自动处理所有工具调用

问题 13: 解释 LangChain 的 LCEL (LangChain Expression Language)。它与 LangGraph 是什么关系?

💡 点击查看答案

答案:

LCEL (LangChain Expression Language) 是什么?

LCEL 是 LangChain 的链式调用语法,使用 | 操作符连接组件。

基础语法:

python

# LCEL 语法
chain = prompt | llm | output_parser

# 等价于
def chain(input):
    step1 = prompt.invoke(input)
    step2 = llm.invoke(step1)
    step3 = output_parser.invoke(step2)
    return step3

# 使用
result = chain.invoke({"topic": "LangGraph"})

LCEL 的核心概念:

1. Runnable 接口

所有 LCEL 组件都实现 Runnable 接口:

python

class Runnable:
    def invoke(self, input):
        """同步调用"""
        pass

    async def ainvoke(self, input):
        """异步调用"""
        pass

    def stream(self, input):
        """流式调用"""
        pass

    def batch(self, inputs):
        """批量调用"""
        pass

2. 管道操作符 |

python

# 串联组件
component1 | component2 | component3

# 数据流:
# input → component1 → output1 → component2 → output2 → component3 → final_output

3. 常用组件

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant"),
    ("human", "{question}")
])

# LLM
llm = ChatOpenAI(model="gpt-4")

# Output Parser
output_parser = StrOutputParser()

# 组合
chain = prompt | llm | output_parser

# 使用
result = chain.invoke({"question": "What is LangGraph?"})

LCEL 示例:

示例 1: 简单问答

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
llm = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

joke_chain = prompt | llm | parser

result = joke_chain.invoke({"topic": "programming"})
# "Why do programmers prefer dark mode? Because light attracts bugs!"

示例 2: 带工具的链

python

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny, 22°C"

llm_with_tools = llm.bind_tools([get_weather])

# 链: 输入 → LLM (可能调用工具) → 输出
agent_chain = llm_with_tools | tool_executor | llm | parser

示例 3: 条件分支 (RunnableBranch)

python

from langchain_core.runnables import RunnableBranch

# 根据输入选择不同的链
branch = RunnableBranch(
    (lambda x: "weather" in x["query"], weather_chain),
    (lambda x: "news" in x["query"], news_chain),
    default_chain  # 默认链
)

result = branch.invoke({"query": "What's the weather today?"})

LCEL vs LangGraph:

特性	LCEL	LangGraph
结构	线性管道	图结构
流程控制	简单条件分支	复杂条件路由
循环	不支持	原生支持
状态管理	隐式传递	显式状态
可视化	难以可视化	图可视化
适用场景	简单链式任务	复杂 Agent 系统

对比示例:

任务: 构建一个 Agent,可以多次调用工具

LCEL 实现 (不优雅):

python

# LCEL 不适合这种场景,需要递归或循环
def lcel_agent(input):
    result = llm_with_tools.invoke(input)
    while result.tool_calls:
        tool_results = execute_tools(result.tool_calls)
        result = llm_with_tools.invoke(input + tool_results)
    return result

# 问题:
# 1. 循环逻辑混杂在代码中
# 2. 状态管理困难
# 3. 难以可视化和调试

LangGraph 实现 (清晰):

python

graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END
})
graph.add_edge("tools", "agent")  # 循环

# 优势:
# 1. 循环结构清晰
# 2. 状态显式管理
# 3. 可视化图结构

LCEL 与 LangGraph 的关系:

1. LangGraph 底层使用 LCEL

python

# LangGraph 节点可以是 LCEL 链
def my_node(state):
    chain = prompt | llm | parser
    result = chain.invoke(state["input"])
    return {"output": result}

graph.add_node("process", my_node)

2. LCEL 是 LangGraph 的"组件"

LangGraph (图编排)
├─ Node 1 (LCEL 链)
├─ Node 2 (LCEL 链)
└─ Node 3 (LCEL 链)

3. 互补关系

python

# LCEL: 节点内部的处理逻辑
node_logic = prompt | llm | parser

# LangGraph: 节点之间的流程控制
def node(state):
    result = node_logic.invoke(state["input"])
    return {"output": result}

graph = StateGraph(State)
graph.add_node("node1", node)
graph.add_edge(START, "node1")

何时使用 LCEL vs LangGraph:

使用 LCEL 的场景 ✅:

python

# 1. 简单的线性处理
chain = prompt | llm | parser

# 2. 单次调用
result = chain.invoke(input)

# 3. 不需要状态管理
# 4. 不需要循环
# 5. 不需要复杂路由

示例: 文档总结、翻译、简单问答

使用 LangGraph 的场景 ✅:

python

# 1. 多轮对话
# 2. 工具调用循环
# 3. 复杂决策树
# 4. 需要人工介入
# 5. 需要状态持久化

示例: ReAct Agent、Multi-Agent 系统、复杂工作流

组合使用 (最佳实践):

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# 1. 用 LCEL 定义节点内部逻辑
summarize_chain = (
    ChatPromptTemplate.from_template("Summarize: {text}")
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)

translate_chain = (
    ChatPromptTemplate.from_template("Translate to Chinese: {text}")
    | ChatOpenAI(model="gpt-4")
    | StrOutputParser()
)

# 2. 用 LangGraph 组织流程
class State(TypedDict):
    text: str
    summary: str
    translation: str

def summarize_node(state):
    summary = summarize_chain.invoke({"text": state["text"]})
    return {"summary": summary}

def translate_node(state):
    translation = translate_chain.invoke({"text": state["summary"]})
    return {"translation": translation}

graph = StateGraph(State)
graph.add_node("summarize", summarize_node)
graph.add_node("translate", translate_node)
graph.add_edge(START, "summarize")
graph.add_edge("summarize", "translate")
graph.add_edge("translate", END)

app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 使用
result = app.invoke({"text": "Long document..."})
# 流程: 文档 → 总结 (LCEL) → 翻译 (LCEL) → 结果

关键洞察:

LCEL 是"顺序执行" 🚂
```
A | B | C  →  A() → B() → C()
```
1

LangGraph 是"图执行" 🗺️

A → [决策] → B 或 C → [可能回到 A]

组合使用最强大 💪

LCEL (节点内部) + LangGraph (节点之间)

从 LCEL 迁移到 LangGraph

python

# LCEL
chain = step1 | step2 | step3

# LangGraph
graph.add_node("step1", step1_func)
graph.add_node("step2", step2_func)
graph.add_node("step3", step3_func)
graph.add_edge("step1", "step2")
graph.add_edge("step2", "step3")

第四部分: 实践与应用 (2 题)

问题 14: 设计一个智能客服路由系统的 LangGraph 架构。要求支持意图识别、FAQ 自动回复、知识库查询、人工转接四种路径。

💡 点击查看答案

答案:

系统架构设计:

用户输入
    ↓
[意图分类]
    ↓
  [路由决策]
    ↓
   ┌──────┼──────┬──────┐
   ↓      ↓      ↓      ↓
[FAQ] [知识库] [技术] [人工]
   ↓      ↓      ↓      ↓
   └──────┴──────┴──────┘
             ↓
         [响应生成]
             ↓
           结束

完整实现:

python

from typing_extensions import TypedDict, Literal
from typing import Annotated
from operator import add
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, START, END

# ========== 1. 状态定义 ==========
class CustomerServiceState(TypedDict):
    # 用户信息
    user_id: str
    user_tier: str  # vip/premium/regular

    # 消息历史
    messages: Annotated[list[BaseMessage], add]

    # 分类结果
    intent: str  # greeting/faq/query/complaint/other
    category: str  # technical/billing/product/other
    sentiment: str  # positive/negative/neutral
    urgency: str  # high/medium/low

    # 处理路径
    routing_path: str  # faq/knowledge_base/technical/human
    requires_human: bool

    # 响应
    response: str
    confidence: float

    # 元数据
    timestamp: str
    processing_time: float

# ========== 2. 节点函数 ==========

# 2.1 意图分类节点
def intent_classification_node(state: CustomerServiceState) -> dict:
    """分析用户消息,识别意图、类别、情感、紧急度"""

    user_message = state["messages"][-1].content

    # 使用 LLM 进行分类
    classifier_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个客服消息分类器。分析用户消息,返回 JSON 格式:
        {{
          "intent": "greeting/faq/query/complaint/other",
          "category": "technical/billing/product/other",
          "sentiment": "positive/negative/neutral",
          "urgency": "high/medium/low"
        }}
        """),
        ("human", "{message}")
    ])

    llm = ChatOpenAI(model="gpt-4", temperature=0)
    chain = classifier_prompt | llm

    result = chain.invoke({"message": user_message})

    # 解析结果 (简化,实际应用应该用 output parser)
    import json
    classification = json.loads(result.content)

    return {
        "intent": classification["intent"],
        "category": classification["category"],
        "sentiment": classification["sentiment"],
        "urgency": classification["urgency"]
    }

# 2.2 FAQ 处理节点
def faq_node(state: CustomerServiceState) -> dict:
    """处理常见问题"""

    # FAQ 数据库 (实际应用应该用向量数据库)
    faq_database = {
        "营业时间": "我们的营业时间是周一至周五 9:00-18:00",
        "退货政策": "购买后 7 天内可以无理由退货",
        "配送时间": "一般 3-5 个工作日送达",
        "支付方式": "支持支付宝、微信、银行卡支付"
    }

    user_message = state["messages"][-1].content

    # 简单匹配 (实际应用应该用语义搜索)
    response = "抱歉,未找到相关 FAQ"
    for key, value in faq_database.items():
        if key in user_message:
            response = value
            break

    return {
        "response": response,
        "confidence": 0.9 if response != "抱歉,未找到相关 FAQ" else 0.3,
        "routing_path": "faq"
    }

# 2.3 知识库查询节点
def knowledge_base_node(state: CustomerServiceState) -> dict:
    """查询知识库 (RAG)"""

    user_message = state["messages"][-1].content

    # 模拟 RAG 流程
    # 1. 向量检索
    retrieved_docs = [
        "产品 X 的技术规格是...",
        "关于计费方式,我们支持...",
    ]

    # 2. LLM 生成答案
    rag_prompt = ChatPromptTemplate.from_messages([
        ("system", """基于以下知识库内容回答用户问题:
        {context}

        如果知识库中没有相关信息,请说明并建议转人工客服。
        """),
        ("human", "{question}")
    ])

    llm = ChatOpenAI(model="gpt-4")
    chain = rag_prompt | llm

    response = chain.invoke({
        "context": "\n".join(retrieved_docs),
        "question": user_message
    })

    return {
        "response": response.content,
        "confidence": 0.8,
        "routing_path": "knowledge_base"
    }

# 2.4 技术支持节点
def technical_support_node(state: CustomerServiceState) -> dict:
    """处理技术问题"""

    user_message = state["messages"][-1].content

    # 调用专门的技术支持 LLM (可能有特殊提示或工具)
    tech_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是专业的技术支持工程师。
        分析用户的技术问题,提供详细的解决方案。
        如果问题复杂,建议转人工工程师。
        """),
        ("human", "{problem}")
    ])

    llm = ChatOpenAI(model="gpt-4")
    chain = tech_prompt | llm

    response = chain.invoke({"problem": user_message})

    # 检查是否需要人工
    requires_human = "转人工" in response.content or state["urgency"] == "high"

    return {
        "response": response.content,
        "requires_human": requires_human,
        "confidence": 0.7,
        "routing_path": "technical"
    }

# 2.5 人工转接节点
def human_handoff_node(state: CustomerServiceState) -> dict:
    """转接人工客服"""

    # 收集所有上下文信息
    context = {
        "user_id": state["user_id"],
        "tier": state["user_tier"],
        "intent": state["intent"],
        "category": state["category"],
        "sentiment": state["sentiment"],
        "urgency": state["urgency"],
        "conversation": [m.content for m in state["messages"]]
    }

    response = f"""已为您转接人工客服。
    工单编号: {state['user_id']}-{state['timestamp']}
    预计等待时间: {"立即接入" if state['user_tier'] == 'vip' else "3-5 分钟"}
    """

    return {
        "response": response,
        "routing_path": "human",
        "requires_human": True
    }

# 2.6 响应生成节点
def response_generation_node(state: CustomerServiceState) -> dict:
    """生成最终响应"""

    # 添加 AI 响应到消息历史
    return {
        "messages": [AIMessage(content=state["response"])]
    }

# ========== 3. 路由函数 ==========

def route_after_classification(state: CustomerServiceState) -> Literal["faq", "knowledge_base", "technical", "human", "end"]:
    """根据分类结果路由"""

    # 规则 1: VIP 用户直接转人工
    if state["user_tier"] == "vip":
        return "human"

    # 规则 2: 投诉 + 负面情绪 → 人工
    if state["intent"] == "complaint" and state["sentiment"] == "negative":
        return "human"

    # 规则 3: 高紧急度 → 人工
    if state["urgency"] == "high":
        return "human"

    # 规则 4: 问候 → 直接结束
    if state["intent"] == "greeting":
        return "end"

    # 规则 5: FAQ → FAQ 节点
    if state["intent"] == "faq":
        return "faq"

    # 规则 6: 技术问题 → 技术支持
    if state["category"] == "technical":
        return "technical"

    # 规则 7: 其他查询 → 知识库
    if state["intent"] == "query":
        return "knowledge_base"

    # 默认: 结束
    return "end"

def route_after_processing(state: CustomerServiceState) -> Literal["response", "human"]:
    """处理后的路由: 检查是否需要转人工"""

    # 置信度低 → 转人工
    if state["confidence"] < 0.5:
        return "human"

    # 明确标记需要人工
    if state.get("requires_human", False):
        return "human"

    # 正常响应
    return "response"

# ========== 4. 构建图 ==========

graph = StateGraph(CustomerServiceState)

# 添加节点
graph.add_node("classify", intent_classification_node)
graph.add_node("faq", faq_node)
graph.add_node("knowledge_base", knowledge_base_node)
graph.add_node("technical", technical_support_node)
graph.add_node("human", human_handoff_node)
graph.add_node("respond", response_generation_node)

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

# 分类后的条件路由
graph.add_conditional_edges(
    "classify",
    route_after_classification,
    {
        "faq": "faq",
        "knowledge_base": "knowledge_base",
        "technical": "technical",
        "human": "human",
        "end": END
    }
)

# FAQ 处理后检查是否需要转人工
graph.add_conditional_edges(
    "faq",
    route_after_processing,
    {
        "response": "respond",
        "human": "human"
    }
)

# 知识库查询后检查
graph.add_conditional_edges(
    "knowledge_base",
    route_after_processing,
    {
        "response": "respond",
        "human": "human"
    }
)

# 技术支持后检查
graph.add_conditional_edges(
    "technical",
    route_after_processing,
    {
        "response": "respond",
        "human": "human"
    }
)

# 所有路径最终都到 END
graph.add_edge("respond", END)
graph.add_edge("human", END)

# 编译
app = graph.compile()

# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# ========== 5. 使用示例 ==========

# 测试用例 1: FAQ
result1 = app.invoke({
    "user_id": "user123",
    "user_tier": "regular",
    "messages": [HumanMessage("你们的营业时间是?")],
    "timestamp": "2024-10-30 10:00:00"
})

print("测试 1 - FAQ:")
print(f"路径: {result1['routing_path']}")
print(f"响应: {result1['response']}")
print()

# 测试用例 2: 技术问题
result2 = app.invoke({
    "user_id": "user456",
    "user_tier": "premium",
    "messages": [HumanMessage("我的软件崩溃了,无法启动")],
    "timestamp": "2024-10-30 10:05:00"
})

print("测试 2 - 技术问题:")
print(f"路径: {result2['routing_path']}")
print(f"响应: {result2['response'][:100]}...")
print()

# 测试用例 3: VIP 用户
result3 = app.invoke({
    "user_id": "user789",
    "user_tier": "vip",
    "messages": [HumanMessage("我需要咨询一个问题")],
    "timestamp": "2024-10-30 10:10:00"
})

print("测试 3 - VIP 用户:")
print(f"路径: {result3['routing_path']}")
print(f"响应: {result3['response']}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359

架构亮点:

多维度分类: 意图、类别、情感、紧急度
智能路由: 基于规则 + 动态决策
置信度检查: 低置信度自动转人工
VIP 优先: 高价值用户快速通道
可扩展: 易于添加新的处理节点

生产环境增强:

python

# 1. 添加监控
def monitoring_node(state):
    log_metrics({
        "user_tier": state["user_tier"],
        "intent": state["intent"],
        "routing_path": state["routing_path"],
        "confidence": state["confidence"],
        "requires_human": state.get("requires_human", False)
    })
    return {}

# 2. 添加缓存
def cached_faq_node(state):
    cache_key = hash(state["messages"][-1].content)
    if cache_key in faq_cache:
        return faq_cache[cache_key]
    result = faq_node(state)
    faq_cache[cache_key] = result
    return result

# 3. 添加 A/B 测试
def ab_test_router(state):
    if state["user_id"] % 2 == 0:
        return "strategy_a"
    return "strategy_b"

关键洞察:

状态驱动路由: 基于多个维度做决策
渐进式降级: 自动 → 半自动 → 人工
用户分层: 不同用户不同服务策略
可观测性: 每一步都记录路径和置信度

问题 15: 总结 Module 1 的核心要点。如果只能记住 5 个最重要的概念,你会选择哪些?为什么?

💡 点击查看答案

答案:

如果只能记住 5 个概念,我会选择:

1. 图状态机思维 (核心哲学) 🎯

是什么: LangGraph 将 Agent 建模为状态机,执行过程是状态的连续转换。

为什么重要: 这是从"希望 AI 工作"到"确保 AI 工作"的根本转变。

记住这个:

传统: 调用 → 黑盒 → 希望得到正确结果
LangGraph: 状态 → 节点 → 新状态 → 可控路由 → 可预测结果

应用: 设计任何 Agent 时,先问:

需要哪些状态?
状态如何转换?
转换的条件是什么?

2. State = 数据 + Channels + Reducers (技术核心) 📊

是什么:

python

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add]  # Channel + Reducer
    #         ^^^^^^^^ 数据类型
    #                  ^^^^^^^^^^^^^^^^^^^^ 更新规则

为什么重要: State 设计决定了:

Agent 能记住什么
节点间如何通信
数据如何累积

记住这个:

State 是"共享内存"
Channels 是"独立通道"
Reducers 决定"合并策略" (覆盖 vs 追加)

常见错误:

python

# ❌ 忘记 add reducer
messages: list  # 会被覆盖!

# ✅ 正确
messages: Annotated[list, add]  # 会追加

3. Node = State → State (函数式思维) ⚙️

是什么: 节点是纯函数,输入状态,输出更新。

python

def node(state: State) -> dict:
    # 1. 读取状态
    data = state["field"]

    # 2. 处理逻辑
    result = process(data)

    # 3. 返回更新 (部分更新!)
    return {"field": result}

为什么重要:

节点是可测试的单元
节点是可复用的组件
节点组合成复杂系统

记住这个:

节点签名: (State) → dict
不是: (State) → State  ← 不要返回完整状态

最佳实践:

一个节点做一件事
纯函数,无副作用
只返回改变的字段

4. Conditional Edge = 动态路由 (控制流的灵魂) 🚦

是什么: 根据状态动态决定下一个节点。

python

def router(state) -> Literal["path_a", "path_b"]:
    if state["score"] > 0.8:
        return "path_a"
    return "path_b"

graph.add_conditional_edges("source", router, {
    "path_a": "node_a",
    "path_b": "node_b"
})

为什么重要: 这是 Agent 智能的体现:

不同情况走不同路径
支持循环 (ReAct)
实现复杂决策树

记住这个:

没有条件边 = 流水线 (固定流程)
有条件边 = Agent (动态决策)

应用模式:

ReAct: agent → tools → agent (循环)
错误处理: try → [success/retry/fail]
用户分流: classify → [vip/regular/...]

5. Human-in-the-Loop = 可控性 (生产环境的必需品) 👤

是什么: 在关键节点暂停,等待人工决策。

python

app = graph.compile(
    checkpointer=memory,  # 必需
    interrupt_before=["critical_action"]
)

# 执行到断点暂停
result = app.invoke(input, config)

# 人工审查后继续
result = app.invoke(None, config)

为什么重要:

安全: 防止 AI 做危险操作
可靠: 人工兜底,提高准确性
合规: 某些场景法律要求人工审批

记住这个:

实验室 AI: 完全自动
生产级 AI: 自动 + 人工审批

应用场景:

金融: 大额交易审批
医疗: 诊断建议确认
法律: 合同审查
运维: 危险操作确认

为什么是这 5 个?

1. 覆盖完整性 🎓

图思维 (哲学)
  ↓
State (数据)
  ↓
Node (处理)
  ↓
Conditional Edge (控制流)
  ↓
Human-in-the-Loop (可控性)

2. 理论 + 实践平衡 ⚖️

理论: 图思维、State 设计
实践: Node 编写、Edge 路由、HITL 实现

3. 从简单到复杂的路径 📈

Level 1: 理解 State 和 Node (基础)
Level 2: 使用 Conditional Edge (进阶)
Level 3: 实现 HITL (生产级)

4. 最高 ROI 💰 这 5 个概念:

占用 20% 的学习时间
解决 80% 的实际问题
是所有高级特性的基础

其他重要但可以后学的概念:

Persistence (持久化): 重要,但基于 State 理解
Streaming (流式输出): 体验优化,非核心逻辑
Sub-graphs (子图): 模块化,建立在图理解之上
Multi-Agent (多智能体): 高级应用,依赖基础概念

学习检验:

如果你真正理解这 5 个概念,你应该能:

✅ 设计一个 Agent 的 State Schema
✅ 写出至少 3 个节点函数
✅ 实现一个条件边路由
✅ 在关键节点添加 Breakpoint
✅ 解释为什么 LangGraph 比线性 Chain 更强大

最终洞察:

LangGraph 不是一个框架,而是一种思维方式。
当你开始用"状态"、"节点"、"边"的语言思考问题时,你就掌握了 LangGraph 的精髓。
从这 5 个概念出发,你可以构建任意复杂度的 Agent 系统。

🎯 学习效果评估

完成这 15 个问题后,请诚实地评估自己:

基础理解 (问题 1-5)

[ ] 我能解释为什么选择 LangGraph
[ ] 我理解图思维 vs 链式思维
[ ] 我知道 Human-in-the-Loop 的应用场景

核心概念 (问题 6-10)

[ ] 我能画出 State、Node、Edge 的关系图
[ ] 我理解 Channels 和 Reducers
[ ] 我能写出完整的 LangGraph 应用模板
[ ] 我能实现条件边路由
[ ] 我理解如何实现循环

LangChain 基础 (问题 11-13)

[ ] 我理解 Messages 类型体系
[ ] 我能定义和使用 Tools
[ ] 我理解 LCEL 与 LangGraph 的关系

实践应用 (问题 14-15)

[ ] 我能设计一个完整的 Agent 架构
[ ] 我能总结核心要点

如果有超过 3 个未选中,建议:

🔄 重新学习相关章节
💻 动手实现对应的代码示例
🤝 在社区寻求帮助

📚 延伸阅读

想要更深入理解 Module 1 的内容?推荐以下资源:

LangGraph 官方教程: https://langchain-ai.github.io/langgraph/tutorials/
ReAct 论文: 理解 Reasoning + Acting 循环
LangChain Academy: 官方课程 Module 0-1

💬 结语

恭喜你完成 Module 1 的学习和复习!

这 15 个问题涵盖了 LangGraph 的核心知识。如果你能流畅回答其中 12 个以上,说明你已经建立了扎实的基础。

记住:

理解 > 记忆: 理解概念比记住 API 更重要
实践 > 理论: 动手写代码比看文档更有效
总结 > 积累: 定期回顾比一次性学习更持久

准备好进入 Module 2 了吗?

在 Module 2 中,我们将学习:

更复杂的 Agent 架构模式
记忆管理和持久化
生产级部署

➡️ 进入 Module 2: LangGraph 核心模式

Module 1 复习撰写者一位相信"主动回顾是最好的学习方法"的教育者2024 年 10 月

小结和复习: LangGraph 框架基础 ​

📚 术语表 ​

📚 本章核心知识回顾 ​

🎯 复习问答 (15 题) ​

第一部分: 框架理解与选择 (5 题) ​

问题 1: LangGraph 的核心设计理念是什么?它解决了传统 Agent 框架的哪些问题? ​

问题 2: 在以下场景中,应该选择哪个框架?请说明理由。 ​

问题 3: LangGraph 相比 CrewAI 的最大优势是什么?什么情况下这个优势至关重要? ​

问题 4: 什么是"图思维"?它与传统的"链式思维"有什么本质区别? ​

问题 5: LangGraph 如何实现 Human-in-the-Loop?举例说明其应用场景。 ​

第二部分: 核心概念 (5 题) ​

问题 6: 解释 LangGraph 中 State、Node、Edge 的关系。用一个类比说明它们如何协同工作。 ​

问题 7: 什么是 Channels?它在状态管理中扮演什么角色? ​

问题 8: 写出一个完整的 LangGraph 应用的最小代码模板,并解释每个部分的作用。 ​

问题 9: 条件边 (Conditional Edge) 的路由函数有什么要求?如何处理多分支路由? ​

问题 10: 在 LangGraph 中,如何实现循环?举例说明一个需要循环的真实场景。 ​

第三部分: LangChain 基础 (3 题) ​

问题 11: 解释 LangChain 中的 Messages 类型。为什么 LangGraph 中经常使用 messages: list[BaseMessage] 作为状态字段? ​

问题 12: 什么是 Tools?如何在 LangGraph 中使用工具?解释 bind_tools 的作用。 ​

问题 13: 解释 LangChain 的 LCEL (LangChain Expression Language)。它与 LangGraph 是什么关系? ​

第四部分: 实践与应用 (2 题) ​

问题 14: 设计一个智能客服路由系统的 LangGraph 架构。要求支持意图识别、FAQ 自动回复、知识库查询、人工转接四种路径。 ​

问题 15: 总结 Module 1 的核心要点。如果只能记住 5 个最重要的概念,你会选择哪些?为什么? ​

🎯 学习效果评估 ​

📚 延伸阅读 ​

💬 结语 ​

小结和复习: LangGraph 框架基础

📚 术语表

📚 本章核心知识回顾

🎯 复习问答 (15 题)

第一部分: 框架理解与选择 (5 题)

问题 1: LangGraph 的核心设计理念是什么?它解决了传统 Agent 框架的哪些问题?

问题 2: 在以下场景中,应该选择哪个框架?请说明理由。

问题 3: LangGraph 相比 CrewAI 的最大优势是什么?什么情况下这个优势至关重要?

问题 4: 什么是"图思维"?它与传统的"链式思维"有什么本质区别?

问题 5: LangGraph 如何实现 Human-in-the-Loop?举例说明其应用场景。

第二部分: 核心概念 (5 题)

问题 6: 解释 LangGraph 中 State、Node、Edge 的关系。用一个类比说明它们如何协同工作。

问题 7: 什么是 Channels?它在状态管理中扮演什么角色?

问题 8: 写出一个完整的 LangGraph 应用的最小代码模板,并解释每个部分的作用。

问题 9: 条件边 (Conditional Edge) 的路由函数有什么要求?如何处理多分支路由?

问题 10: 在 LangGraph 中,如何实现循环?举例说明一个需要循环的真实场景。

第三部分: LangChain 基础 (3 题)

问题 11: 解释 LangChain 中的 Messages 类型。为什么 LangGraph 中经常使用 `messages: list[BaseMessage]` 作为状态字段?

问题 12: 什么是 Tools?如何在 LangGraph 中使用工具?解释 `bind_tools` 的作用。

问题 13: 解释 LangChain 的 LCEL (LangChain Expression Language)。它与 LangGraph 是什么关系?

第四部分: 实践与应用 (2 题)

问题 14: 设计一个智能客服路由系统的 LangGraph 架构。要求支持意图识别、FAQ 自动回复、知识库查询、人工转接四种路径。

问题 15: 总结 Module 1 的核心要点。如果只能记住 5 个最重要的概念,你会选择哪些?为什么?

🎯 学习效果评估

📚 延伸阅读

💬 结语