企业官网建设流程全解析

做一个能真正“跑起来”的 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 Language（LCEL），也能把几个组件链式组合起来。但 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 token，5 轮之后就到万级了。我通常会在agent节点调用模型前做一次消息截断，只保留最近几轮，或者用摘要模型把旧对话压成一段概述。以 8K 上下文窗口为例，我用的保守策略是保留最近 10 条消息，超出部分触发一轮总结。

3. 源码实战：从零实现一个客服工单 Agent

3.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，逐个执行并把结果包装成ToolMessage：

def 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(content=str(result), tool_call_id=tool_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(checkpointer=MemorySaver())

这个图的逻辑非常直白，但又和手写循环有本质区别：你可以在compile之后对图做持久化、做 State 回放、按thread_id恢复上下文。比如用户中途离开，几天后回来继续问，只要传入同一个thread_id，Agent 能接上之前的对话状态。

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}]}, config=config ) 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}]}, config=config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"] token = chunk.content or "" if token: yield f"data: {token}\n\n" return StreamingResponse(event_generator(), media_type="text/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 给模型看，模型大概率会尝试换一种方式调用或给出说明，整个链路不至于白跑。

这次从架构选型到源码落地的完整过程，我最大的体会是：Agent 未来真正拉开差距的，一定不是模型本身有多聪明，而是工程化能力能不能兜住模型的不可靠。LangGraph 最大的价值不是“让代码更炫”，而是让 Agent 的行为变得可观测、可控制、可恢复。在它之前，我调试 Agent 基本靠玄学；在它之后，我至少能明确回答“刚才下一步去哪了”这个问题。

最后分享一个小技巧：我给节点命名时会刻意用“动作+对象”的语义化命名，比如agent、tools、validate_input。光看名字就知道这个节点在干什么，调试日志和追踪可读性会好一个量级。如果你正在从单次 Prompt 调用往真正 Agent 方向走，建议先别急着堆功能，把状态图里的流转逻辑画清楚再动手写代码，能省下大量瞎折腾的时间。