Agent 笔记

基于 deeplearning.ai 公开课整理的 LangGraph 学习笔记，覆盖 State 管理、Tool Calling、 ReAct 循环、记忆系统、Human-in-the-Loop、Multi-Agent 与 RAG 集成。
注：以下内容在 AI 辅助下整理。核心概念的理解、代码示例和实践总结经过了人工校对。部分 API 行为可能随 LangGraph 版本更新有变化，请以官网文档为准。

第一章LangGraph 基础概念

State、Node、Edge、Tool Calling、Checkpointer 与 Streaming

1. 为什么学 LangGraph

在接触 LangGraph 之前，我对 Agent 的理解就是简单的 LLM 调用：扔一个 prompt，拿回结果，完事。这在单轮问答场景下没问题，但一旦涉及多轮对话、工具调用、条件分支，这种简单模式就撑不住了。

接触 LangGraph 后，最大的感受是：它把 Agent 的工作流变得显式可读了。Agent 不只是"调 LLM + 解析输出"，而是一个多步骤的工作流——每一步有明确的输入输出，步骤之间有条件跳转，中间状态需要被保存和追踪。

对我来说，LangGraph 解决了一个核心问题：如何让 Agent 的行为可控、可追踪、可中断。这在涉及工具调用和多步推理的场景中尤为关键。

2. 我对 LangGraph 的整体理解

我目前的理解是：LangGraph 是一个把 Agent 应用组织成"带状态的图"的框架。三个概念就能概括：

State（状态）：在所有节点间共享的数据字典，承载消息历史、工具结果和中间推理步骤。
Node（节点）：一个处理单元。比如"调用 LLM"是一个节点，"执行工具"是另一个节点。
Edge（边）：节点之间的跳转规则。普通边是固定跳转，条件边则根据当前 State 决定下一步去哪。

打个比方：如果 Agent 是一个"自动化办事流程"，State 就是那张流转中的表格，Node 是各个办事窗口，Edge 就是连接窗口的箭头。这个结构让复杂流程变得可管理。

3. State：Agent 的共享工作台

State 不是普通的 Python 变量。它最关键的特性是：每个节点返回的部分状态会和全局 State 合并（merge），而不是覆盖。不同节点更新不同字段，互不干扰。

在多轮对话中，State 通常包含：

messages：完整的对话历史（用户消息、AI 回复、工具调用、工具结果）
next_step：供条件边判断路由方向
自定义字段，比如 context、intermediate_steps

LangGraph 中使用 TypedDict 或 Pydantic 模型定义 State。我目前用 TypedDict + Annotated，因为它可以为每个字段指定 reducer 逻辑（比如消息字段用 add_messages 来累积而不是覆盖）。

注意：add_messages 的具体行为（如同 ID 消息的更新逻辑）在不同 LangGraph 版本间可能有细微差异。

4. Node 和 Edge：把流程拆成可控的步骤

在用 LangGraph 之前，我写 Agent 逻辑的方式是一个大函数吃到底：调 LLM → 解析 → 需要就调工具 → 再解析 → 返回。简单场景能跑，但问题很明显：

✕ 所有逻辑搅在一起，加一个步骤就要改主函数
✕ 出问题时很难判断是"LLM 判断错了"还是"工具执行错了"
✕ 没有复用性，"调工具"这段逻辑每次都要重写

LangGraph 的 Node + Edge 模式解决了这个问题。每个节点是一个单一职责的函数：

# LLM Node
def call_model(state: AgentState) -> dict:
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

# Tool Node
def call_tools(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    tool_results = []
    for tool_call in last_message.tool_calls:
        result = execute_tool(tool_call)
        tool_results.append(result)
    return {"messages": tool_results}

条件边负责路由决策：

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "__end__"

graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "__end__": "__end__"
})

5. Tool Calling：让模型从「生成文本」变成「调用能力」

这是学 LangGraph 最让我开眼界的部分。LLM 只能输出文本，但 Tool Calling 把它从"回答者"变成了"编排者"——它不执行，而是判断需要什么、然后请求执行。

完整的工具调用流程：

1LLM 接收用户输入，判断是否需要调用工具

2如果需要，LLM 不输出文本，而是输出 tool_calls（包含工具名和参数）

3条件边检测到 tool_calls，路由到工具执行节点

4工具节点实际执行工具，将结果以 ToolMessage 形式追加

5流程回到 LLM 节点，LLM 结合工具结果生成最终回答

关键认知：模型不执行工具——它请求执行。实际的工具执行发生在程序代码中。这把"决策"和"执行"分开了，意味着工具可以是一切：数据库查询、API 调用、脚本执行，甚至启动另一个 Agent。

6. Checkpointer：为什么 Agent 需要记住状态

一开始我很困惑：对话之间的状态怎么持久化？没有持久化，每次新的用户输入都从零开始，之前的工具结果和推理全都丢了。

Checkpointer 解决了这个问题，它像游戏的存档系统：

每个 thread_id 对应一个独立的对话线程
每经过一个图步骤，State 自动存档（checkpoint）
下次用同一个 thread_id 调用，从上次存档点恢复

目前我开发调试只用内存 MemorySaver。如果要上线，需要换成持久化后端（SqliteSaver 或 PostgresSaver）。

7. Streaming：为什么流式输出很重要

graph.stream() 是我觉得最实用的功能之一。除了渐进式输出带来的用户体验提升，它对我更大的价值是调试可见性。

通过流式输出，你能看到：

Agent 每一步执行了哪个 Node
LLM 输出了什么（包括 tool_calls）
工具调用的参数和返回值
条件边的路由决策

没有流式输出时，我靠 print 语句追踪流程。有了流式，每一步都清清楚楚，调试效率大幅提升。把这个过程暴露给前端，也能让用户看到 Agent 的"思考过程"，增加信任感。

8. 一个最小工作流例子

一个最小的 LangGraph 工作流，展示了 ReAct 循环的核心逻辑：

from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[Sequence, add_messages]

def agent_node(state: AgentState):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_msg = state["messages"][-1]
    results = [execute(tc) for tc in last_msg.tool_calls]
    return {"messages": results}

def router(state: AgentState):
    last_msg = state["messages"][-1]
    if getattr(last_msg, "tool_calls", None):
        return "tools"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", router)
builder.add_edge("tools", "agent")
graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
    {"messages": [HumanMessage(content="weather?")]},
    config
)
for event in events:
    print(event)

如果 LLM 判断需要天气工具，流程是：agent → tools → agent → END。如果不需要：agent → END。整个过程逐步流式输出，Agent 的决策链一目了然。

9. 我的阶段性理解

它不是聊天封装，是工作流引擎。LangGraph 把 Agent 的每一步（调用模型、执行工具、条件跳转）显式组织成图，流程可控可追踪。
State 是核心。理解了 State 的 merge 机制和 reducer 设计，基本就理解了 LangGraph 的一半。
图和条件边让流程显式化。Agent 不再是黑盒——每一步、每个分支都可追踪。
Tool Calling 不神秘。它就是一个结构化的"请求-执行-反馈"循环，LLM 只做判断。
我目前处于"能用"阶段，离"用好"还有距离。复杂路由场景、异常处理、多工具时的性能——这些都还在学习中。

← 已是第一章1 / 8