尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

AI Agent工程落地指南:七要素、关键决策与LangGraph实现

AI Agent工程落地指南:七要素、关键决策与LangGraph实现 过去三个月我一直在折腾 AI Agent 的工程落地。从前端到后端从提示词到状态恢复把一个从零搭起来的“会聊天”的 Demo逐步改成了“能干活”的小系统。所谓能干活指的是它能听懂用户一句话自己规划要查哪个接口、调用哪几个工具把结果整理成可读回复甚至在某个环节中断之后还能接上进度继续跑完。如果你也准备上手做 Agent或者已经写了几个月提示词但总觉得离生产环境差一口气这篇笔记应该能帮你把拼图补完整。我不打算写论文式的定义只讲真实工程视角下的 Agent先拆成七个构成要素再列七个我每次搭建都必须做出选择的决策点最后给一段基于 FastAPI 和 LangGraph 的最小可运行示例能抄作业就直接抄。1. AI Agent 的七要素全景图动手写代码前先画一张分工图我最早看各种 Agent 框架的文档最大的收获其实是一张草图中心是大模型周围围着一圈记忆、工具、行动、反馈。后来在项目里踩了大量坑才发现很多问题不是模型不够聪明而是边界没划分清楚该由哪个模块负责的事被混在了一起。1.1 先看全局哪七个要素缺一不可先给出结论。在我看来所有 Agent 无论用什么框架、什么协议、什么部署方式都可以拆成七个要素目标定义、感知输入、记忆、规划、工具、行动执行、反思。这七个要素不是代码层面的模块名而是问题域的划分。比如感知输入在不同场景里可能是 Webhook 回调、文件读取、图片解析、数据库查询你没法用一个统一的“感知模块”包打天下但你必须在设计时回答“这个 Agent 能感知什么、不能感知什么”。我把七个要素的职责和核心反问整理成了一张表要素核心职责设计时必须回答的问题目标定义把用户需求转成可执行的任务约定Agent 到底要完成什么不做什么感知输入获取并标准化外部信息输入来源有哪些格式怎么归一化记忆保存和检索上下文、业务事实、中间结果哪些信息要短期保留哪些要长期沉淀规划将大任务拆成小步骤决定执行顺序步骤是固定的还是动态生成的依赖关系怎么表达工具扩展模型能力边界每个工具的参数、返回、错误、幂等性是否清晰行动执行与环境交互并产生真实副作用执行失败怎么处理副作用能不能回滚反思检查结果、识别错误、触发修复怎么判断一次执行是成功还是需要重试这七样东西少任何一样短期可能靠提示词和经验值硬撑但一遇到线上流量、异常输入、外部依赖抖动就会暴露问题。1.2 输入侧三要素目标、感知、记忆如何协同目标定义是第一道坎也是最容易被一句提示词带过的地方。很多人把“目标”直接写在 System Prompt 里比如“你是一个智能客服要快速响应用户”但到了执行层面“快速”和“严格按知识库回答”可能产生冲突。我踩过的典型坑是目标写得含糊模型开始自由发挥自作主张编造接口参数。正确的做法是在设计阶段就把目标拆成约束条件比如“只调用白名单工具”“非确定信息必须标注不确定”“答案必须包含依据来源”让模型有明确的判据而不是靠感觉。感知输入决定了 Agent 的信息质量。文本输入相对简单但生产环境里更多是图片、扫描件、表格、网页抓取结果。我的经验是感知层不要指望模型用长上下文去硬扛应该在做入口处做预处理。比如 PDF 先转成文本再进模型表格先做列名映射网页先去掉导航噪音。否则模型会被无关内容干扰工具调用时也会拿错字段。感知设计得越干净后面的规划压力越小。记忆这件事很多人误以为就是“把历史消息全塞给模型”。实际上至少要分三层短期记忆是当前会话的消息列表长期记忆是跨会话的用户画像和业务事实工作记忆是任务执行过程中产生的临时中间结果。三者不能混在一个列表里。短期记忆要考虑长度上限长期记忆需要检索而不是全量加载工作记忆要和状态管理打通。设计顺序上我建议先做长期记忆因为短期记忆可以被上下文窗口覆盖长期记忆才真正需要向量检索或键值存储来支撑。1.3 执行侧四要素规划、工具、行动、反思如何闭环规划可以简单到一条 ReAct 规则“先想、再动、再看结果”也可以复杂到用另一层 LLM 调用把任务拆成一组带依赖关系的 DAG。以我实测的经验超过一定复杂度之前不要让模型做重量级规划。用一个白名单动作集加条件分支通常更稳也更省钱。真正值得动态规划的场景是任务步骤事先无法确定或者外部返回会影响下一步走向。比如查天气这种固定任务写死流程就好但“帮我订一个符合要求的行程”这种任务才需要模型拆解成查景点、查交通、查住宿多个步骤。工具是 Agent 的毛细血管每接一个工具都要想清楚四件事参数 Schema 是什么、返回结构长什么样、错误语义怎么表达、是否幂等。其中幂等性最容易被忽略。用户点一次重试和系统自动重试如果产生两个订单、两条消息那问题就不是模型笨而是接口设计有缺陷。我在项目里吃过这个亏后来要求所有写操作工具必须支持 request_id 去重重试才敢放开。行动执行看起来只是发一个 HTTP 请求或者跑一段函数但它藏着三个容易被低估的问题工具副作用能否回滚调用超时怎么办返回数据不合法怎么办。这些问题在设计阶段就要写清楚而不是等运行时报错再补救。反思则是闭环的最后一环也最容易被砍掉。我建议至少保留一层轻量校验模型每次调用工具后对返回结果做合法性判断发现不对就换一种方式重试。实测下来很多“Agent 死循环”就是因为少了这个简单校验模型拿到错误结果像没头苍蝇一样反复调用同一个工具。2. 七个决策点真正决定 Agent 工程成败的地方要素是静态的认知框架决策点是动态的工程选择。我在项目立项时会强迫自己回答七个问题并把答案写进设计文档。这些问题没有标准答案但早做决定一定比晚做决定省事。2.1 决策一主模型怎么选不是越贵越好主模型选型是第一个要拍板的事。我的建议是不要只盯着跑分榜单重点看三件事工具调用成功率、结构化输出稳定性、长上下文下的信息保持能力。工具调用成功率直接决定了 Agent 能不能准确把意图映射到函数参数结构化输出稳定度决定了后续解析代码要不要写一堆兜底逻辑长上下文保持能力决定了几十轮对话后模型会不会把早期关键信息丢掉。成本与延迟也要分场景。实时客服类 Agent 对延迟敏感适合便宜且快的小模型离线任务型 Agent 可以接受几十秒的思考时间可以考虑更强推理能力的模型。我的经验是日常工具调用场景选择一个中端模型比如 gpt-4o-mini、DeepSeek、Qwen 系列就够用只有多跳推理、复杂代码生成、长文档分析才需要换更强模型。结论是分层使用不是一套模型打天下。2.2 决策二提示词工程怎么设计顺序比辞藻重要提示词怎么写直接影响 Agent 的稳定性。我的顺序一直是先写目标再写边界然后写行为公约接着写工具说明最后写输出格式。不要把大量笔墨花在人格化描述上模型会把精力消耗在模仿语气上反而忽略任务本身。系统提示词追求的是“一句话能讲清约束”而不是文笔优美。如果工具数量多不要把工具说明全部塞进提示词。主流模型都支持 Function Schema 绑定提示词里只需要写使用策略和禁忌工具名称和参数描述交给模型平台的函数调用能力去处理。这个区分很重要我见过有人把几十个工具的手写文档全放进 System Prompt结果上下文被挤爆模型反而开始幻觉。此外要防指令注入工具返回的内容可能包含外部信息要在提示词里明确要求模型把“数据”和“指令”区分开不可执行的文本一律视为数据。2.3 决策三记忆层怎么分层存储不能只靠一个列表记忆实现上我建议分两层热数据和冷数据。热数据用 Redis存放最近几轮对话的原文、用户 session 状态、任务中间变量冷数据用向量数据库存放跨会话需要检索的长期记忆和业务知识。目前我用得比较顺手的组合是 pgvector 加 Redispgvector 在 PostgreSQL 里做向量检索少维护一个组件Redis 负责热路径缓存。上下文膨胀是个必现问题。当 Token 快满时不要等模型报错应该做一个滚动摘要动作把早期对话交给一个小模型压缩成摘要再拼上最近几轮原文继续作为上下文输入。这个动作可以定时触发也可以依据 token 统计触发。我在生产环境里把它做成了一个独立节点效果比把历史全部塞给主模型好得多因为主模型可以一直保持在它最擅长的推理工作区而不是被迫去 processing 一堆老对话。2.4 决策四工具调用走哪个协议功能调用还是 MCP工具调用的接入方式有两种主流选择。第一种是各模型供应商原生支持的 Function Calling。它把工具描述以 JSON Schema 形式传给模型模型生成结构化的调用请求实现简单走通一条链路只要半天。缺点是它和供应商平台绑定切换模型时可能要做适配。第二种是 MCPModel Context Protocol这类标准化协议它把工具能力变成可发现、可复用的服务适合中大型项目或需要跨多个服务共享工具集的场景。我对选型的经验是MVP 阶段直接用 Function Calling最快验证业务闭环当工具数量超过十几个或者工具由不同团队维护时再迁移到 MCP。不要一开始就上标准化协议除非团队里已经有多套服务要接入 Agent否则学习成本大于收益。当然如果项目当前就在做面向多Agent的工具市场这种基础设施那 MCP 从第一天就要考虑进去。2.5 决策五任务编排用哪种模式不是所有场景都需要 Agent任务编排模式的选择我总结成一句话能用确定性流程解决的问题别上 Agent真需要动态决策的环节加 ReAct需要跨步骤共享状态和失败恢复的直接用有状态图。线性 Chain 适合流水线式处理比如“输入 - 清洗 - 分类 - 输出”稳定高效但没有应变能力。ReAct 适合“查一下、观察结果、再查”这类动态循环典型场景是搜索引擎问答。Plan-and-Execute 则先规划再执行适合任务步骤较多、但每个步骤相对明确的场景。LangGraph 这类图编排方案则把任务建模成节点和边支持条件分支、循环、人工审批和状态持久化几乎能覆盖上面所有模式。我的建议是先按真实业务场景确定编排模式再挑框架而不是看着框架的宣传语倒推业务架构。实践中我常用混合结构主流程用图编排图内个别需要推理的节点用 ReAct既保证可控又保留灵活性。编排方式适合场景优点主要缺点Chain输入到输出的固定流水线稳定、调试简单无法处理分支与回退ReAct需要观察中间结果的问答/搜索灵活、贴近模型天然推理易陷入反复循环、Token 开销大Plan-and-Execute任务可预先拆解但步骤较多全局规划更清晰规划有错时整体失败率高Graph多任务、多状态、需要恢复的复杂流程可控性最强、支持持久化概念复杂开发量稍大2.6 决策六并发与性能怎么设计Agent 扛并发的问题在哪“AI Agent 怎么扛并发”是很多人关心的问题但它和普通 Web 服务扛并发有本质区别。你的服务本身可能很轻真正的瓶颈几乎都在上游 LLM API 的 RPM 和 TPM 限制上。如果不对上游做保护服务一扩容限流就会立刻响应接着就是请求堆积、超时重试、重试放大了流量最后雪崩。所以 Agent 并发架构的第一要务是保护上游而不是压满自己的吞吐。我的基本架构是四件套接口层全部异步化用 FastAPI 的 async def 处理 HTTP 请求遇到同步的 LangGraph 调用丢进线程池避免阻塞事件循环入口加令牌桶限流按用户维度控制请求速率任务量大的场景把请求放入 Redis 队列由 Worker 异步消费客户端轮询结果需要体验更好时用 SSE 做流式输出边生成边推送给用户。另外相似请求可以做结果缓存比如热点问题直接命中缓存根本不打大模型。有一个特别容易踩的坑重试策略。无脑重试会把一次限流变成持续的重试风暴。正确做法是指数退避加抖动并且区分可重试错误429、超时和不可重试错误参数非法、认证失败。我在项目里就是靠这个区分把线上错误率降了一个数量级。2.7 决策七安全与可观测性让 Agent 可控可查让大模型自己决定调用所有工具是一件很危险的事情。权限控制的核心是最小权限每个 Agent 只能使用完成当前任务所必需的工具集合写操作工具必须增加人工审批闸门。比如删除数据、发送消息、下单支付这类操作流程上做成“Agent 生成意图人工点击确认后才真正执行”。这既是对用户的保护也是对自己系统的保护。可观测性同样重要。每一个节点的输入输出都应该打快照尤其是模型调用、工具调用和分支跳转。我建议从第一天就接入 trace 系统无论是 LangSmith、Langfuse 还是自建日志链路都要保证能回答三个问题某次任务走到了哪个节点、模型视角看到了什么、工具真实返回了什么。没有 traceAgent 就是一个黑盒出了问题只能靠猜有了 trace大部分问题都能快速定位到是决策问题还是数据问题。3. 最小可落地示例用 FastAPI LangGraph 搭一个带记忆的 Agent理论部分讲了这么多最后必须落到代码上。我用一个极简但完整的最小示例展示一条能跑通的 Agent 链路带工具调用、带记忆恢复、通过 HTTP 暴露服务。技术栈是 FastAPI 加 LangGraph因为前者是异步友好的 Python Web 框架后者把状态管理和条件分支做成了显式建模非常贴近七要素里的规划与反思。3.1 技术栈选型为什么是 FastAPI 和 LangGraph选择 FastAPI 是因为它天然支持 async配合异步 HTTP 客户端和 SSE 流式输出在 Agent 场景里比 Flask 和 Django 更顺手。选择 LangGraph 不是因为它最流行而是它解决了其他方案很难处理的问题状态管理。普通 ReAct 实现需要自己维护循环和上下文LangGraph 则把状态、节点、边、检查点都做成了一等公民尤其是检查点机制让我可以用极少的代码做到用户断线后重连继续任务。这个示例我会绑定一个模拟天气查询工具模型收到“北京天气怎么样”这类问题时会生成一次工具调用工具节点执行完后把结果返回给模型模型再组织语言回复。整个流程只有两个节点但已经覆盖了七要素里的大部分内容目标定义在系统提示词里、感知输入就是用户消息、工具和行动在同一个节点内、反思由条件边实现记忆通过 Checkpointer 保存。3.2 核心代码状态图、工具调用和检查点# app.py import asyncio from typing import TypedDict, Annotated from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver tool def get_weather(city: str) - str: 查询指定城市的当前天气。 # 这里用模拟数据实际项目中替换成真实天气 API return f{city} 当前天气晴最高气温 28 摄氏度。 llm ChatOpenAI(modelgpt-4o-mini, temperature0).bind_tools([get_weather]) class AgentState(TypedDict): messages: Annotated[list, add_messages] def call_model(state: AgentState): result llm.invoke(state[messages]) return {messages: [result]} def call_tool(state: AgentState): last_message state[messages][-1] if not last_message.tool_calls: return {messages: []} tool_call last_message.tool_calls[0] if tool_call[name] get_weather: content get_weather.invoke(tool_call[args]) else: content 未知工具 return { messages: [ { role: tool, name: tool_call[name], tool_call_id: tool_call[id], content: content, } ] } graph StateGraph(AgentState) graph.add_node(model, call_model) graph.add_node(tool, call_tool) graph.add_edge(tool, model) graph.add_conditional_edges( model, lambda state: tool if state[messages][-1].tool_calls else END, {tool: tool, END: END}, ) graph.set_entry_point(model) # MemorySaver 只能在单机内存中保存状态生产环境建议换成 Redis 或 PostgreSQL 实现 compiled_agent graph.compile(checkpointerMemorySaver()) app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str default app.post(/chat) async def chat(request: ChatRequest): config {configurable: {thread_id: request.thread_id}} output await asyncio.to_thread( compiled_agent.invoke, {messages: [HumanMessage(contentrequest.message)]}, config, ) return {reply: output[messages][-1].content}这段代码有几个关键点值得展开。bind_tools把工具的 JSON Schema 自动传给模型这是决策四里说的 Function Calling 思路。add_messages是一个消息缩减器LangGraph 在更新 state 时会把新旧消息正确合并避免每次覆盖。MemorySaver是检查点实现它会在每次节点执行后保存状态即使中断也能用同一个thread_id恢复上下文。asyncio.to_thread的用法值得说明一下compiled_agent.invoke是同步阻塞调用直接放在 async 函数里会阻塞事件循环。用to_thread把它丢到线程池执行HTTP 层才能继续并发处理其他请求。这是 FastAPI 集成 LangGraph 时常犯的一个错误我在这里提前排掉。3.3 本地跑起来启动、调用、验证多轮记忆pip install fastapi uvicorn langchain-openai langgraph export OPENAI_API_KEYyour-api-key uvicorn app:app --host 127.0.0.1 --port 8000然后打开另一个终端curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 北京天气怎么样, thread_id: user-123}第一次调用会看到模型先触发工具调用然后返回类似“北京当前天气晴最高气温 28 摄氏度”的回复。接着你再发一次curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 那上海呢, thread_id: user-123}因为用了同一个thread_id第二次请求会自动带上第一次的对话历史模型能理解“那上海呢”指的是查询上海天气。这就是 MemorySaver 带来的多轮记忆能力。如果把thread_id换成一个新值就是一段全新对话两条会话互不打扰。4. 常见问题与排查实录把故障变成经验库任何 Agent 项目上线后都会遇到一批重复性故障。这里把我在实际项目中反复排查过的五个问题整理成一个速查表再分享一些设计阶段的避坑心得。4.1 高频故障现场五个真实案例复盘故障现象根因分析快速解决方案Agent 反复调用同一个工具形成死循环没有对工具返回做合法性校验模型拿到错误结果后继续重试在调用工具后增加结果校验节点判断返回是否包含预期字段对话进行到一半出现 Token 超限上下文只增不减没有做滚动摘要在主循环里增加摘要节点对话达到阈值时先压缩再继续并发一上来就报上游 429 限流没有限流和重试策略请求全打到模型服务加入令牌桶限流对 429 使用指数退避加重试用户断开后重新连接上下文丢失检查点没有持久化状态停留在内存把 MemorySaver 换成 Redis / PostgreSQL 实现的后端工具返回了结构正确的数据但模型用错字段工具返回 Schema 不够显式模型靠猜在工具返回中增加字段说明或使用严格 JSON Schema 解析第一个案例特别典型。模型在对话中生成了一个工具调用工具返回了一段格式不匹配的内容模型没有意识到失败又在下一个推理步骤里重复生成几乎一样的工具调用整个流程卡死。解决办法是增加一个“结果校验节点”在进入下一轮模型调用之前检查返回内容是否满足预期结构不满足就重新构造工具结果并说明错误原因逼模型换路径。Token 超限的问题在长对话场景里几乎必现尤其做客服和写作助手时。滚动摘要不是一次性把所有历史都删掉而是把早期对话用一个小模型压缩成几条要点再拼接最近两三轮的原文。这样既能保留关键信息又能把主模型的上下文空间留出来处理最新任务。4.2 避坑心得清单设计阶段就避开的雷复盘多个项目之后我总结了几个设计阶段的实用心得。第一不要把全部逻辑押给模型。Agent 能动态决策不等于所有环节都要动态。固定流程用代码写死动态决策才交给模型这是最省钱也最稳定的组合。第二工具的错误语义必须明确规定。返回值要区分“调用失败”“无结果”“数据不合法”三类情况模型才能依据错误类型决定重试还是终止。第三Trace 从第一天就接上。哪怕是个人项目也要把每次模型调用、工具调用的输入输出记录下来这些数据不光用于排查问题也用于优化提示词和工具定义。第四不要在日志里打 API Key哪怕本地调试也要养成好习惯。第五先跑通一条最小链路再扩展工具集。我第一次做 Agent 时就是一口气接了一堆工具结果问题叠着问题根本分不清是哪个环节出错。最后再分享一个习惯我自己做任何新 Agent 项目时都会先用七要素画一张草图再把七个决策点的答案填进一张表格然后才开始写代码。这套流程帮我省掉了大量返工也推荐你试试。折腾 Agent 的过程很像搭乐高框架看起来谁都会真正的差别在于你愿不愿意在每个细节上多想一层边界和异常。先想清楚目标和边界再让它跑起来这条经验在任何 Agent 项目里都适用。
返回列表