AI AgentTechnical Deep Dive

第一个 LangGraph Agent:先跑通最小闭环,再谈复杂编排

发布时间2026/01/10
分类AI Agent
预计阅读10 分钟
作者吴长龙
*

第一次用 LangGraph,不要一上来就做“全能助手”。先把状态、工具、循环和记忆跑通,后面的复杂流程才有工程基础。

01.第一个 LangGraph 项目,别从“功能很多”开始

很多入门教程会把第一个 Agent 写成一个大而全的演示:

  • 能搜网页
  • 能算数学
  • 能记上下文
  • 能多轮循环
  • 还能分支路由

看起来很热闹,但对刚上手的人不一定友好。真正容易让你理解 LangGraph 的,不是功能数量,而是下面这条闭环有没有跑通:

  • 模型判断下一步
  • 工具执行并回写结果
  • 图根据结果继续或结束
  • 同一线程能在下一次调用时接着跑

只要这四件事清楚了,后面再加更多工具、更多节点、人工审批和长期记忆,才不会越写越乱。

02.LangGraph 真正解决什么问题

根据 LangGraph 当前官方文档,它的定位不是“高层 Agent 一键封装”,而是一个面向长运行、可恢复、带状态的 Agent 编排框架。它强调的核心能力包括:

  • durable execution,也就是中断后还能继续
  • human-in-the-loop,也就是流程里可以插入人工决策
  • memory,也就是线程级和跨会话的状态持久化
  • observability,也就是让每一步的状态和路径可追踪

这意味着 LangGraph 更适合处理下面这类问题:

  • 一个任务会经历多个步骤和多次工具调用
  • 任务中途可能失败、暂停、等待确认
  • 系统需要保留状态和历史轨迹
  • 你希望显式控制节点、边和路由逻辑

如果你只是做一个简单问答助手,LangChain 的高层 agent 往往更省力。LangGraph 的价值主要体现在“流程复杂而且需要控制力”的地方。

03.先做一个最小闭环

下面这个示例有意保持克制。它只做一件事:如果用户提问需要天气信息,就调用工具;否则直接回答。

第一步:定义工具和模型

python snippetpython
from langchain.chat_models import init_chat_model
from langchain.tools import tool


model = init_chat_model("openai:gpt-4.1-mini", temperature=0)


@tool
def get_weather(city: str) -> str:
    """查询指定城市天气。"""
    sample = {
        "北京": "晴,25C",
        "上海": "多云,23C",
    }
    return sample.get(city, "未找到该城市天气信息")


tools = [get_weather]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

这里最重要的不是模型本身,而是工具定义足够明确。对 LangGraph 来说,工具越小、越单一、越像稳定接口,整条链路越容易调试。

第二步:定义状态

python snippetpython
import operator
from typing_extensions import Annotated, TypedDict
from langchain.messages import AnyMessage


class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

这个状态已经够用:

  • messages 保存整条会话里的消息
  • llm_calls 帮助我们观察循环有没有失控

官方 quickstart 也采用类似写法,用 reducer 把新消息追加到原有状态,而不是直接覆盖。

第三步:定义模型节点和工具节点

python snippetpython
from langchain.messages import SystemMessage, ToolMessage


def call_model(state: AgentState) -> dict:
    response = model_with_tools.invoke(
        [
            SystemMessage(content="需要实时信息时调用工具,信息不足时不要猜。")
        ] + state["messages"]
    )
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }


def call_tools(state: AgentState) -> dict:
    results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        results.append(
            ToolMessage(
                content=observation,
                tool_call_id=tool_call["id"],
            )
        )

    return {"messages": results}

这个分层很关键:

  • 模型节点只负责判断和生成 tool call
  • 工具节点只负责执行工具并返回结果

不要把“问模型”和“执行外部动作”揉成一个函数。拆开以后,日志、重试和回放都会更清晰。

第四步:定义循环和退出条件

python snippetpython
from typing import Literal
from langgraph.graph import StateGraph, START, END


def should_continue(state: AgentState) -> Literal["call_tools", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tools"
    return END


builder = StateGraph(AgentState)
builder.add_node("call_model", call_model)
builder.add_node("call_tools", call_tools)
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue, ["call_tools", END])
builder.add_edge("call_tools", "call_model")

graph = builder.compile()

到这里,我们已经有了一个最小 LangGraph Agent:

  • 入口从 STARTcall_model
  • 如果模型发出了 tool call,就转到 call_tools
  • 工具结果写回后,再回到 call_model
  • 如果没有 tool call,就走到 END

这就是最典型的“模型判断 -> 工具执行 -> 模型收敛”的循环。

04.给第一个 Agent 加上记忆

LangGraph 的真正优势之一,是它原生支持线程级状态持久化。官方文档里推荐通过 checkpointer 来保存每一步的状态。

入门阶段可以先用内存版:

python snippetpython
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage


checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "demo-user-1"}}

graph.invoke(
    {"messages": [HumanMessage(content="北京天气怎么样?")]},
    config,
)

graph.invoke(
    {"messages": [HumanMessage(content="那上海呢?")]},
    config,
)

这里的 thread_id 很重要。只要线程 ID 一致,LangGraph 就能把新的输入接到同一条会话状态上。官方文档也明确把这类能力归到 short-term memory,也就是线程级记忆。

如果进入生产环境,就不该继续使用内存 checkpointer,而应该换成数据库后端,让状态在服务重启后仍然能恢复。

05.第一个实战最容易踩的坑

1. 把状态当 prompt 文本缓存

状态里应该放“原始数据”,比如消息、工具结果、分类结果,而不是提前拼好的长 prompt。官方文档在 “Thinking in LangGraph” 里也强调过这一点:状态应尽量保持原始,prompt 在节点里按需格式化。

这样做的好处是:

  • 不同节点可以各自组织上下文
  • 更容易调试
  • 更容易演进状态结构

2. 工具职责过大

如果你定义一个 “万能工具” 去做搜索、数据库查询和外部写操作,模型很容易选不准,也很难追踪错误来源。

更稳的做法是:

  • 一个工具只做一个动作
  • 参数尽量少
  • 返回结构尽量稳定

3. 没有显式退出策略

虽然图里已经有 END,但实际项目里通常还要补更多边界,例如:

  • 最大 LLM 调用次数
  • 工具失败重试上限
  • 高风险动作强制人工确认

如果这些边界不在第一版就考虑,后面功能一多,图很容易变成“能跑但不可控”。

06.什么时候该继续往下加复杂度

当你已经能稳定跑通上面的最小闭环,再考虑下面这些能力会更合适:

  • Command 把状态更新和路由写在同一个节点里
  • interrupt() 做人工审核和恢复
  • 为易失败节点加 RetryPolicy
  • 把检索、分类、执行拆成多个子图
  • 接入数据库 checkpointer 做真正的断点恢复

顺序不要反过来。先搭清楚最小骨架,再上复杂流程,学习曲线会平缓很多。

07.参考资料