做一个能真正“跑起来”的 AI Agent和做一个能完美回答“什么是 Agent”的 PPT完全是两码事。纸上谈兵聊 ReAct、聊记忆、聊工具调用聊得再顺都不如亲手把一个 Agent 的源码跑通来得踏实。今天这篇东西就是我从零开始用 LangGraph 编排 Agent 内部流程、再用 FastAPI 把它包装成 HTTP 服务的完整记录。不吹概念不扯大词直接用代码和架构讲清楚 AI Agent 在工程上到底是怎么落地的。适合已经写过几段 Prompt、跑过基础 RAG Demo现在想往工程化方向走一走的开发者。这篇不是入门教程而是踩过坑之后的反刍笔记。1. 为什么是 LangGraph FastAPI 这个组合1.1 AI Agent 的工程本质链式逻辑不够用了很多人对 Agent 的理解就是“调用一个大模型让它自己发挥”。如果只是做一个聊天机器人那确实一个model.invoke()就够了。但 Agent 和聊天机器人的本质区别在于Agent 需要在对话过程中根据用户的意图主动决定调用什么工具、按什么顺序处理、拿到结果之后如何反馈给用户。这就意味着控制逻辑不是一条直线而是一个循环。最典型的循环就是 ReAct模型思考Reason→ 决定调用工具Act→ 观察工具结果Observe→ 再思考……直到模型认为问题已经解决输出最终答案。如果把这个逻辑用 Python 硬写也不是不行while max_steps 0: output llm.invoke(messages) if output.tool_calls: messages.extend(execute_tools(output.tool_calls)) else: return output.content max_steps - 1十几行代码就能跑通一个最基础的循环。但是一旦你开始处理多工具、多步骤、多轮记忆、中断恢复、并发生成这些实际问题这种手写 while 循环就会迅速膨胀成一座意大利面条山每一步的状态不可记录、不可恢复、不可并行。我一开始也是这么干的写到最后基本是靠“祈祷它能跑”线上一个问题排查半天找不到状态到底在哪一步出了问题。1.2 编排框架选型LangGraph 的定位LangGraph 解决的核心问题就是把 Agent 的控制流从“一坨循环代码”变成“一张有向图”。你需要声明节点Node和边Edge节点是具体的计算步骤边决定控制流的走向。Agent 的行为逻辑被显式地画出来每个节点都能访问一份全局共享的状态State状态的变化可追踪、可回放、可持久化。其实 LangChain 早期也有LangChain Expression LanguageLCEL也能把几个组件链式组合起来。但 LCEL 适合的是“固定管线”比如检索→提示词→模型→输出解析器管线一旦确定就不太会变。而 Agent 的流程是动态的模型每一步的输出都可能改变下一步走向这种动态路由用状态图表达比线性链表达自然得多。从我实际使用的体验来对比LangGraph 和传统手写循环的区别就像从“用手捏泥巴”变成“用模具做零件”。泥巴灵活但形状不可控模具前期有学习成本但成型之后无论是调试、测试、复用都要稳得多。1.3 API 层选型FastAPI 的优势图已经把 Agent 内部编排好了但用户不会直接用 Python 对象跟你交互必须把 Agent 暴露成 HTTP 服务。选 FastAPI 几乎是顺理成章的事它有四个优势对我这种后端场景非常关键原生异步支持。Agent 推理本身是慢操作加上工具调用可能长达十几秒异步 IO 能保证 API 层在等待模型输出时不会阻塞其他请求。Streaming 响应支持。Agent 生成过程是流式的FastAPI 配合StreamingResponse可以很好地做 SSE 逐字输出用户不用干等着。Pydantic 请求校验。入参、出参定义成模型之后接口文档自动生成调用方也不容易传错参数。生态成熟。部署到常见的容器平台、对接监控告警都非常顺。和 Flask、Django 对比过之后Flask 偏轻但异步支持一般Django 适合重业务但为了一个 Agent 服务背上整套框架不划算。FastAPI 是那种“刚拿起来觉得顺手跑生产也撑得住”的选择。2. 核心概念与关键算法拆解2.1 StateGraph把 Agent 的循环变成有向图LangGraph 里最核心的对象是StateGraph。先定义一份 State——通常用TypedDict声明然后往里挂节点、连边。我建议第一次接触的人先抓住三个概念State图上所有节点共享的“口袋”节点读它、改它、返回它。Node一个 Python 函数接收 State返回对 State 的部分更新。Edge节点之间的连接可以是无条件跳转也可以是条件路由。这里有一个特别需要注意的设计细节节点不要直接修改 State 里的字段而是把修改后的值作为返回值返回。我最早写代码时习惯在函数内部直接state[messages].append(...)结果状态更新有一搭没一搭。LangGraph 的架构是“被图执行框架收到的返回值才会合并进 State”你在函数内部改了外面的字典图管理的内存状态感知不到严格来说这不是刻板的框架限制而是为了保证更新的可追踪性和并发安全。State 里的字段还有一个可选行为叫 Reducer。最常用的场景是messages字段如果多个节点都想往消息列表里追加内容直接用普通 list 会导致后写的覆盖先写的。用add_messages这个 Reducer 就能实现“追加”而不是“覆盖”from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next: str我拿它类比“多人共用的工作台”普通 list 相当于每个人进场都把工作台清空再放自己的东西加了一个累加型 Reducer 之后每个人只是往台面上再叠一块积木谁叠了什么都有迹可循。2.2 ReAct 循环与工具调用算法Agent 的“智能感”很大程度上来自工具调用。实现工具调用的前提是模型支持 function calling。简单说就是在请求模型时除了系统提示词和对话历史再附带一份 JSON Schema 格式的工具描述列表。模型在生成过程中可以输出一段结构化内容表示“我要调用某个工具参数是什么”而不是直接输出普通文本。LangGraph 把这件事包装成了节点之间的路由算法。我的实现里定义了两个核心节点agent节点把当前消息列表发给模型模型返回一个带tool_calls的 AIMessage。tools节点解析上一步的tool_calls在真实代码里执行对应的工具函数把结果包装成 ToolMessage 追加进 State。边采用的是条件路由如果模型返回内容里有tool_calls路由到tools节点如果没有说明模型准备直接给最终答复路由到结束节点。执行完工具之后工具结果会作为新消息回到agent节点模型看到工具结果后再做一次决策循环往复。这个循环的终止条件主要有两种模型不再输出tool_calls或者达到最大迭代次数。我在生产环境里两种都会加尤其最大迭代次数不然后端可能出现一个 Agent 循环十几轮停不下来的情况既烧 token 又拖垮响应时间。简化后的路由判断代码长这样def route_agent(state: AgentState): if state.get(next) tools: return tools return END注意这里的next字段是agent节点在返回时设置的不是模型输出里自带的。这样做在调试时非常有用因为你从 State 里一眼就能看到当前流程走到了哪一步。2.3 记忆机制短期记忆、长期记忆与检查点Agent 要处理多轮对话记忆是一个绕不开的话题。LangGraph 的记忆体系分两层短期记忆存在于当前 State 里的对话消息序列模型上下文的直接来源。长期记忆跨会话保留的信息通常落到外部存储向量数据库、Redis、Postgres 等。其中短期记忆的持久化靠 Checkpointer。LangGraph 的MemorySaver是最简单的内存版 Checkpointer适合开发调试。生产环境我用过基于 Redis 和 Postgres 的实现核心作用就一个把每一轮运行结束后的 State 快照保存下来下次用同一个thread_id调用时就能接着上一次的状态继续对话。这里容易踩的坑是 Context 爆炸。模型上下文窗口有限我的经验是每轮对话大约消耗 1500 到 2500 token5 轮之后就到万级了。我通常会在agent节点调用模型前做一次消息截断只保留最近几轮或者用摘要模型把旧对话压成一段概述。以 8K 上下文窗口为例我用的保守策略是保留最近 10 条消息超出部分触发一轮总结。3. 源码实战从零实现一个客服工单 Agent3.1 准备工作与项目结构我用一个业余时间做的“客服工单 Agent”作为 Demo 来讲。需求很常见用户描述问题Agent 判断是否需要查订单、是否需要走退款流程、能不能直接答复。项目结构我按职责拆成四块agent_service/ ├── requirements.txt ├── app/ │ ├── main.py # FastAPI 入口 │ ├── agent_graph.py # LangGraph 图定义 │ ├── tools.py # 工具函数定义与注册 │ └── schemas.py # Pydantic 请求响应模型依赖安装很简单我用的是 Python 3.11 环境pip install langgraph langchain-openai fastapi uvicorn pydantic如果你用的是其他模型服务商只要它兼容 OpenAI 的 function calling 协议驱动封装层换成对应实现就行图结构完全不用改。这一点 LangGraph 做得比较好底层模型和上层编排是解耦的。3.2 定义状态与工具先定义 State。我把messages和next都设计成必填字段next用来显式记录路由目标class AgentState(TypedDict): messages: Annotated[list, add_messages] next: str工具函数保持简单语义化。一个查订单一个查退款状态一个提交退款请求。注意工具名一定要用“动词宾语”的语义化命名因为模型是靠名字和描述理解该在什么场景下调用哪个工具的命名含糊模型很容易选错。def get_order_status(order_id: str) - str: 查询订单当前状态 # 模拟查询实际生产替换成数据库或外部 API 调用 return f订单 {order_id} 当前状态已发货预计 3 天内送达 def get_refund_status(order_id: str) - str: 查询退款进度 return f订单 {order_id} 退款申请处理中预计 2 个工作日到账 def submit_refund(order_id: str, reason: str) - str: 提交退款申请 return f订单 {order_id} 退款申请已提交原因{reason}每个工具函数的 docstring 就是模型看到的功能描述所以别随便写两行要认真写清参数含义、返回格式和典型调用场景。3.3 构建 Agent 工作流核心图构建我拆成两部分。先在agent_graph.py里定义两个节点函数call_model负责调用模型。这里我加了一个关键逻辑传给模型的messages先截断到最近 10 条这一步是防止上下文膨胀的兜底策略。模型返回的AIMessage如果带tool_calls就设置next为tools否则设置为end。def call_model(state: AgentState) - dict: recent_messages state[messages][-10:] response llm_with_tools.invoke(recent_messages) if getattr(response, tool_calls, None): return {messages: [response], next: tools} return {messages: [response], next: end}execute_tools节点遍历上一条 AIMessage 中的tool_calls逐个执行并把结果包装成ToolMessagedef execute_tools(state: AgentState) - dict: last_message state[messages][-1] tool_results [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_func tools_by_name[tool_name] result tool_func.invoke(tool_args) tool_results.append( ToolMessage(contentstr(result), tool_call_idtool_call[id]) ) return {messages: tool_results}然后构建图。START是虚拟入口点条件边从agent节点出发根据next字段决定跳转到tools还是END。tools执行完之后无条件回到agent让模型看到工具结果并继续决策。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, execute_tools) builder.add_edge(START, agent) builder.add_conditional_edges( agent, lambda state: state.get(next) or end, {tools: tools, end: END} ) builder.add_edge(tools, agent) graph builder.compile(checkpointerMemorySaver())这个图的逻辑非常直白但又和手写循环有本质区别你可以在compile之后对图做持久化、做 State 回放、按thread_id恢复上下文。比如用户中途离开几天后回来继续问只要传入同一个thread_idAgent 能接上之前的对话状态。3.4 FastAPI 集成与流式输出FastAPI 这层只做三件事接收请求、调用图、返回结果。先定义一个请求体class AgentRunRequest(BaseModel): thread_id: str message: str普通非流式接口app.post(/agent/run) def run_agent(req: AgentRunRequest) - dict: config {configurable: {thread_id: req.thread_id}} result graph.invoke( {messages: [{role: user, content: req.message}]}, configconfig ) return {reply: result[messages][-1].content}这是最简单的一版但对于耗时十几秒的 Agent 任务用户体验很差。我后来改成了 SSE 流式输出让模型生成的每个 token 都能陆续到前端from fastapi.responses import StreamingResponse app.post(/agent/stream) async def stream_agent(req: AgentRunRequest): config {configurable: {thread_id: req.thread_id}} async def event_generator(): async for event in graph.astream_events( {messages: [{role: user, content: req.message}]}, configconfig, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk] token chunk.content or if token: yield fdata: {token}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)这里要提一个细节SSE 和浏览器的EventSource有个老问题——EventSource只支持 GET 请求而我们需要 POST 请求。我的方案是前端用fetch接口自己读取响应流手动解析 SSE 格式的数据兼容性完全没问题。3.5 运行与测试启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000我用一段简单的 curl 验证流程是否正常。先问一个不涉及工具的问题curl -X POST http://localhost:8000/agent/run \ -H Content-Type: application/json \ -d {thread_id: 1, message: 你好我想查一下订单 1024 的状态}预期返回结果会经过一次工具调用最后回到模型生成的自然语言答复。如果模型判断需要工具State 路由会经历agent → tools → agent → END从日志里能看到完整的流转轨迹。我再强调一下temperature的调参心得Agent 场景和写文案不同模型输出最好是稳定、可预期。工具调用决策阶段温度调到 0最终答复生成阶段可以稍微提高一点但别超过 0.3不然模型容易犯错——明明该调工具却自作主张编一个答案。4. 常见问题与排查技巧实录4.1 状态被“只读参数”坑了我最早碰到过一种诡异情况节点函数内部打印 State 时明明看到messages增加了但下一轮节点又看不到。排查到最后发现问题出在“直接改参数没把更新结果返回”。LangGraph 的设计是节点函数只接收 State所有更新必须以返回值的形式交给框架合并。如果发现自己对 State 做了修改但图里不生效优先检查是否返回了dict且键名与 State 定义完全一致。messages字段有没有加add_messagesReducer。有没有在节点内部忘记return。这条规则一开始觉得绕习惯之后反而舒服因为每个节点变成纯函数单测非常容易写。4.2 流式响应与 Agent 循环互相等待整套流程跑通之后我把响应方式从完整 JSON 改成流式结果前端等了十几秒才看到第一个字。原因是图内部做了多轮工具调用而我只在最后一步封装了流式输出中间的工具调用过程用户完全感知不到。LangGraph 的astream_events会把所有事件打出来包括模型推理和工具执行。我需要过滤出真正想要的on_chat_model_stream事件并且把工具调用阶段的“中间态”以提示信息形式发给前端比如在工具节点里 yield 一条data: [system] 正在查询订单状态...。这样用户等待时会看到进度而不是一片空白。4.3 FastAPI 阻塞同步图卡住异步事件循环这个坑非常隐蔽。FastAPI 是异步框架如果直接在一个async def接口里调用 LangGraph 的同步invoke方法会阻塞整个事件循环并发来两个请求时第二个会一直排队。我压测时才暴露出来症状是 QPS 上不去某个请求对象长时间占用。解法有两种用graph.ainvoke(...)或astream_events(...)走异步接口。如果必须用同步图把它放到线程池里执行await asyncio.to_thread(graph.invoke, ...)。我生产环境选择直接统一用异步接口LangGraph 的异步支持很完善没有理由回到同步路线上。4.4 记忆膨胀与上下文爆炸长会话跑久了 token 计数会一路飙升最终模型开始“忘记”用户最初的诉求。这不是模型笨而是上下文中无关信息太多注意力被稀释了。我采取的方案是分层处理固定窗口策略只保留最近 N 条消息这适合大多数轻量客服场景。摘要压缩策略每满 10 轮对话做一次总结把历史内容压缩成一段摘要替换掉旧窗口。这适合要保留关键信息的业务场景代价是每轮多一次摘要模型调用。外部向量库把历史消息向量化后存储按相关性动态召回。这套适合复杂业务但对小项目来说太重了我暂时没上。先推荐大部分读者用第一种简单可靠不会引入新的失败点。4.5 工具返回格式不合法某个工具函数返回了一个 Python 字典对象我直接str(result)转成了字符串模型看起来吃力还有一次工具内部抛异常没捕获整个工具节点直接中断Agent 流程就崩了。我的兜底原则是工具函数内部必须捕获所有异常返回统一的错误文本返回内容尽量是结构化文本最好能让模型直接引用。比如{status: success, data: ...}这种 JSON 文本比一句含糊的自然语言可靠得多。如果工具执行失败不要中断流程把错误原因原样封装成 ToolMessage 给模型看模型大概率会尝试换一种方式调用或给出说明整个链路不至于白跑。问题典型症状原因解决方案状态更新丢失节点读不到上一节点写入的消息直接修改参数而非返回更新节点必须返回 dictmessages 用 add_messages流式卡顿用户长时间看不到输出只有最后一步做了流式astream_events 过滤模型事件中间过程给提示并发阻塞第二请求排队同步 invoke 堵住事件循环用 ainvoke / astream_events 或 asyncio.to_threadtoken 膨胀长会话模型跑题历史消息无限制增长固定窗口截断 摘要压缩工具崩溃Agent 流程中断工具内部异常未处理工具函数全量 try/except返回错误文本这次从架构选型到源码落地的完整过程我最大的体会是Agent 未来真正拉开差距的一定不是模型本身有多聪明而是工程化能力能不能兜住模型的不可靠。LangGraph 最大的价值不是“让代码更炫”而是让 Agent 的行为变得可观测、可控制、可恢复。在它之前我调试 Agent 基本靠玄学在它之后我至少能明确回答“刚才下一步去哪了”这个问题。最后分享一个小技巧我给节点命名时会刻意用“动作对象”的语义化命名比如agent、tools、validate_input。光看名字就知道这个节点在干什么调试日志和追踪可读性会好一个量级。如果你正在从单次 Prompt 调用往真正 Agent 方向走建议先别急着堆功能把状态图里的流转逻辑画清楚再动手写代码能省下大量瞎折腾的时间。