学习目标

从手写 Agent 过渡到工业级框架，掌握 LangChain 工具系统和 LangGraph 状态图。

为什么需要 LangGraph？

Module 1 的手写 Agent 有这些问题：

- 手动管理消息列表容易出错

- 状态管理没有规范化

- 没有内置的持久化、流式输出

- 复杂 Agent（多分支、并行）难以实现

LangGraph 将 Agent 建模为一个状态机，解决了以上所有问题。

---

2.1 LangChain 工具抽象 (langchain_tools.py)

三种创建方式

# 方式 1: @tool 装饰器 (推荐，最简洁)
@tool
def calculator(expression: str) -> str:
    """执行数学计算。"""
    ...

# 方式 2: StructuredTool (自定义 schema)
search_papers = StructuredTool.from_function(
    func=search_papers_func,
    name="search_papers",
    description="搜索自动驾驶论文",
)

# 方式 3: BaseTool 子类 (完全控制)
class ReadFileTool(BaseTool):
    name: str = "read_file"
    args_schema: type[BaseModel] = ReadFileInput
    def _run(self, filepath: str) -> str: ...

对比 Module 1

	Module 1 (手写)	Module 2 (LangChain)
工具定义	dataclass + 手动 schema	@tool 装饰器自动生成
LLM 绑定	手动 prompt 拼接	llm.bind_tools(tools)
工具调用	手动 JSON 解析	AIMessage.tool_calls 自动解析

---

2.2 LCEL 链式调用 (simple_chain.py)

LCEL = LangChain Expression Language

核心操作：| (管道符)

# 最简单的 chain
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"role": "专家", "question": "什么是 BEV？"})

# 带结构化输出
chain = prompt | llm | JsonOutputParser(pydantic_object=MyModel)

# 并行执行
parallel = RunnableParallel(
    academic=prompt_aca | llm | parser,
    practical=prompt_pra | llm | parser,
)

---

2.3 StateGraph 入门 (first_graph.py) ⭐

三个核心概念

# State: Agent 的"记忆"
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# Node: 处理函数
def chatbot_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Edge: 状态转移
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

条件路由

# 根据分类结果路由到不同专家
graph.add_conditional_edges(
    "classifier",
    route_by_category,  # 返回 "tech" | "science" | "general"
    {"tech": "tech_expert", "science": "science_expert", "general": "general"}
)

---

2.4 工具调用 Agent (tool_calling_agent.py)

bind_tools -- Agent 开发的核心 API

llm = create_llm()
llm_with_tools = llm.bind_tools(tools)

# LLM 会自动产生 tool_calls
response = llm_with_tools.invoke(messages)
# response.tool_calls = [{"name": "calculator", "args": {"expression": "2+3"}}]

Graph 结构

START → llm_node → [router] → tool_node → llm_node → ...
                       └→ END (无工具调用时)

---

2.5 手动 ReAct Agent Graph (react_agent_graph.py) ⭐ 核心模板

这是整个教程最重要的 Graph 模板。后续 Module 3-7 的所有 Agent 都在此基础上扩展。

手动构建的意义

虽然 LangGraph 提供了 create_react_agent 预构建函数，但手动构建能让你理解：

1. agent_node 做什么（调用 LLM，决定是否用工具）

2. tool_node 做什么（执行工具，返回 ToolMessage）

3. should_continue 做什么（条件路由）

4. State 如何在节点间流动

完整代码结构

def build_react_agent(tools, system_prompt):
    llm_with_tools = create_llm().bind_tools(tools)

    def agent_node(state):   # 大脑: 调用 LLM
        return {"messages": [llm_with_tools.invoke(state["messages"])]}

    def tool_node(state):    # 手脚: 执行工具
        tool_calls = state["messages"][-1].tool_calls
        results = [execute_tool(tc) for tc in tool_calls]
        return {"messages": results}

    def should_continue(state):  # 路由器
        if state["messages"][-1].tool_calls:
            return "tools"
        return END

    builder = StateGraph(State)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", tool_node)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, ...)
    builder.add_edge("tools", "agent")
    return builder.compile()

---

2.6 条件路由 Agent (routing_agent.py)

模式：Supervisor + Specialists

START → supervisor → [router]
                       ├→ ad_expert (自动驾驶工具)
                       ├→ cv_expert  (CV 工具)
                       └→ general    (通用聊天)

适用：客服路由、多领域专家系统、代码审查分类。

---

2.7 Plan-and-Execute (plan_execute.py)

vs ReAct

	ReAct	Plan-Execute
思路	思考→行动→观察→再思考	先规划→逐步执行→汇总
适合	简单短任务	复杂多步任务
可见性	看不到全局计划	用户能看到完整计划

START → planner → executor → [more?] → executor → summarizer → END

---

2.8 流式输出与 Checkpoint (streaming_checkpoints.py)

Streaming

# 实时看到 Agent 每一步做了什么
for event in agent.stream(input, config, stream_mode="updates"):
    print(event)

Checkpoints (断点续聊)

memory = MemorySaver()
agent = builder.compile(checkpointer=memory)

# 同一 thread_id = 共享记忆
config = {"configurable": {"thread_id": "user-123"}}
agent.invoke(msg1, config)  # 第1轮
agent.invoke(msg2, config)  # 第2轮，记住第1轮内容

---

2.9 Human-in-the-Loop (human_in_loop.py)

高风险操作需要审批

agent = builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"],  # 执行工具前暂停
)

result = agent.invoke(...)  # 在 tools 前自动暂停
# 人工审批后继续
agent.invoke(Command(resume={"approved": True}), config)

---

Module 2 核心总结

概念	一行理解
@tool	装饰器自动生成工具 schema
`\	` 管道符	LCEL 的核心语法：`prompt \	llm \	parser`
StateGraph	Agent = 状态机 (State + Nodes + Edges)
bind_tools	让 LLM 能够自动产生 tool_calls
add_conditional_edges	根据 LLM 输出路由到不同节点
MemorySaver	自动持久化，支持断点续聊
interrupt_before	关键操作前暂停等审批