上周四晚上十点我把开发了三周的AI智能体推到生产环境然后在工位上又坐了两个小时盯着监控面板上的请求数慢慢爬上来才终于把心跳按回去。这个智能体不是那种只会陪聊的机器人——它能自己拆任务、调系统里的工具接口把“查一下上个季度订单量”这种自然语言请求转成一次真实的数据库查询和报表生成。今天这篇就把这个AI Agent从立项、开发到上线的完整过程摊开讲一遍。我尽量按可复现的逻辑写包括为什么选这套技术栈、图状工作流怎么编排、工具调用怎么设计、并发上来怎么顶住以及上线后的监控和那些绕不开的坑。适合正在做Agent项目的后端工程师、想入门的毕业生还有想评估“我们自己要不要自研Agent”的技术负责人。1. 项目概述与设计思路1.1 先搞清楚AI Agent和普通聊天机器人差在哪很多团队上来就喊“做个智能体”结果做出来还是一个聊天机器人。区别到底在哪普通聊天机器人是“你说一句我回一句”本质是文本接龙没有目标也不会自主调用外部能力。AI Agent则是一个完整的“感知-决策-执行”闭环你给它一个目标它可以自己规划出好几步每一步去调用一个工具查库存、生成图片、发消息看完工具返回的结果再决定下一步干什么。这个循环在业界有个名字叫ReAct模式就是Reason思考 Act行动交替进行。我用生活类比解释一下像你雇了一个靠谱的临时助理你跟他说“帮我订周五下午到上海的机票”他不会只回一句“好的”而是会先查航班工具1再对比价格工具2然后问你选哪个航班或者直接下单工具3。每做一步都要观察结果、再决定下一步这就是Agent的行为模式。在我这个项目里智能体主要处理三类请求业务查询类订单状态、商品信息、内容生成类营销文案、商品图、事务操作类创建工单、发送通知。业务方对它的要求就一条能把“一句话需求”变成“一串可执行的动作”而不是只会打嘴炮。1.2 需求拆解与技术路线选型动手之前我先把这个项目的需求拆成了四个层面模型层谁来当大脑做推理和决策。编排层怎么控制流程让Agent按逻辑一步步走而不是随机发挥。工具层Agent能调哪些外部接口怎么安全地调。接入层怎么把能力暴露给业务系统让别人方便调用。技术选型方面我实际对比过三套方案这里直接说结论。第一套是扣子这类低代码平台搭建确实快拖拽编排、内置插件不少适合做原型验证和轻业务但要做复杂的鉴权、自定义并发策略、私有化部署就非常难受。第二套是从零手写全套流程最灵活可模型调用的重试、解析、上下文管理、异常恢复全得自己处理开发成本高不说后期维护也痛苦。第三套是LangChain LangGraph FastAPI的组合LangChain提供模型封装和工具调用生态LangGraph提供图状的状态机编排FastAPI负责对外暴露HTTP接口和异步并发处理。我最后选了这套就是看中它在“可控”和“开发效率”之间的平衡。提示选型不是追新框架关键是看“评估-反馈-再评估”这个循环能不能低成本地落到代码里。低代码适合快速试错代码编排适合认真做产品。2. 核心架构与关键技术拆解2.1 模型层让大模型当“大脑”而不是“嘴”模型是Agent的推理中枢选型直接决定你的智能体是聪明还是呆。我的选型标准有四个指令遵循能力强能识别用户意图里隐含的子任务原生支持函数调用Function Calling否则工具层的可靠性要大打折扣上下文长度不能太小不然Agent跑几步下来历史记录就爆了成本和延迟要在可接受范围内。用表格把当时对比的几个选项列一下方便参考模型函数调用上下文窗口相对成本备注GPT-4o系列稳128k高指令遵循很强适合复杂任务DeepSeek系列稳64k / 128k低中文理解好性价比高开源本地模型看部署框架看具体模型中可私有化需要GPU资源系统提示词System Prompt是很多人忽略的关键点这个坑我踩得很深。不要只写“你是智能助手”而要明确角色定义、任务边界、可用工具列表、输出格式、安全约束五要素。比如我明确限定只有用户主动询问订单、物流信息时才能调用订单查询工具绝不臆造订单号。一个精简的System Prompt结构大概是这样角色你是XX电商平台的智能客服助手。 任务边界只处理订单查询、商品咨询、退换货引导三类问题其他问题礼貌转人工。 可用工具query_order、query_product、create_ticket。 输出规范回答必须包含信息来源工具名称无法确认的信息必须如实说明。 安全约束不透露内部系统信息不执行无权限的写操作。2.2 工具定义给Agent装上“手脚”工具定义决定了Agent能力的边界。每个工具在代码里就是一个Python函数加一段JSON Schema描述名字、详细描述、参数列表。这里多说一句大模型是靠工具描述来选择调用哪个工具的所以描述必须写清楚“什么时候用、怎么用、参数怎么填”描述不精确等于给Agent装了个乱来的手脚。举个例子订单查询工具的Schema我是这样写的{ name: query_order, description: 查询订单状态与物流信息。仅在用户询问订单详情、物流进度时使用。, parameters: { type: object, properties: { order_id: { type: string, description: 用户订单号必须来自用户输入或上下文不得编造 } }, required: [order_id] } }工具命名也讲究最好以动词开头比如query_order、generate_image、send_notification语义清晰。返回结构必须稳定我统一用{success: true, data: {...}}包裹这样Agent解析结果时不容易乱。2.3 记忆与上下文管理Agent的“记事本”Agent跑得越久上下文就会越膨胀。你不可能把每一步的工具返回全塞进历史记录不现实成本也顶不住。我把记忆分成两层短期记忆和长期记忆。短期记忆就是对话历史但只保留最近N轮长期记忆把业务侧沉淀的用户偏好、常见问答存到向量数据库按需检索。如果项目还没上向量库最少也要做缓存和摘要。我实际采用的上下文管理策略贴出来直接抄对话轮次超过20轮时只保留最近10轮完整消息更早的做一轮摘要后替换。单次工具返回超过2000字符先让模型压缩成结构化摘要再放回上下文。核心业务参数用户ID、订单号、商品ID用独立字段存储不走对话历史。每条消息记录token数接近阈值时主动触发压缩。这套策略上线后单请求token消耗下降了大概百分之三十效果非常明显。2.4 工作流编排用图把ReAct循环固定下来Agent的核心是循环不是线性调用。模型决定调什么工具执行工具把结果回填给模型模型再决策直到它认为任务完成。LangGraph正是把这种循环建模成图的工具节点是状态更新边是状态转移。我在项目里建了两个节点agent节点负责让模型决定下一步动作tools节点负责执行具体工具再用一条条件边判断是继续循环还是结束。逻辑上它的流程就是用户输入进入agent节点模型分析后如果决定调用工具就进入tools节点执行工具返回结果后回到agent节点再做判断如果模型认为已经完成就输出最终回复。我给这条循环加了一个硬性限制最大轮数设成5步超过直接终止并返回“需要转人工处理”防止死循环白烧token。为什么要用图而不是简单写个while循环因为大型Agent场景会复杂得多有的分支需要人工审批、有的分支要并行调多个工具、有的分支要回退重试。图的表达能力能把这些都覆盖而且每一步的状态都可以追踪和可视化排障时特别有用。3. 实操过程与核心实现3.1 环境搭建一套能跑Agent的基础依赖说一千道一万不如直接看代码。先给环境。我用了Python 3.11虚拟环境管理用venv核心依赖不多也就这几个python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn langchain langchain-openai langgraph redis httpx pytest环境变量统一放到.env文件里管理参考字段如下OPENAI_API_KEYsk-xxx # 如果用DeepSeek则配DEEPSEEK_API_KEY和DEEPSEEK_BASE_URL DATABASE_URLpostgresql://user:passlocalhost:5432/agent REDIS_URLredis://localhost:6379/0 MAX_STEPS5 MODEL_NAMEgpt-4o-mini这里提一句团队如果同时接多个模型商建议把模型调用接口封装成一个Factory类按配置切换而不是散落在代码各处。3.2 核心代码实现一个能查订单的智能体下面是核心代码我做了简化但结构完整能直接跑通。第一步定义工具函数def query_order(order_id: str) - dict: # 实际项目里这里会查数据库或调用订单服务 data find_order_by_id(order_id) if not data: return {success: False, error: f订单 {order_id} 不存在} return {success: True, data: { order_id: order_id, status: data[status], logistics: data[logistics] }}第二步创建模型并绑定工具。LangGraph里通常这样注册from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def search_order(order_id: str) - str: 查询订单状态。仅在用户询问订单信息时调用。 result query_order(order_id) return str(result) llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([search_order])第三步用LangGraph构建Agent图from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated class AgentState(TypedDict): messages: list def agent_node(state: AgentState): # 模型决定下一步回复用户 or 调用工具 response llm_with_tools.invoke(state[messages]) return {messages: [response]} def tools_node(state: AgentState): # 执行模型要求的所有工具调用 last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: tool_result search_order.invoke(tool_call) results.append(tool_result) return {messages: results} def should_continue(state: AgentState): last state[messages][-1] return tools if last.tool_calls else END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.set_entry_point(agent) graph.add_edge(tools, agent) graph.add_conditional_edges(agent, should_continue) app graph.compile()这段代码的核心逻辑就一句话Agent节点决策有工具调用就去执行执行完回来再决策直到不再需要工具为止。我特别强调一下should_continue这个条件边是生死线——每次都判断是否还有工具调用没有就退出循环防止死循环。第四步用FastAPI把它暴露成HTTP接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str message: str class ChatResponse(BaseModel): session_id: str reply: str app.post(/api/agent) async def chat(req: ChatRequest): # 简化的会话初始化逻辑实际按session_id拉取历史 initial_state {messages: [{role: user, content: req.message}]} # 用ainvoke做异步调用这是高并发下不阻塞worker的关键 result await app_graph.ainvoke(initial_state) last_msg result[messages][-1] return ChatResponse(session_idreq.session_id, replylast_msg.content)3.3 高并发AI Agent怎么扛住流量这是上线前团队问得最多的问题也是整个项目最见功夫的地方。先认清两个事实Agent接口天然“慢”一次完整交互往往要多次LLM调用基准耗时能到2到10秒不能用普通接口1秒超时的标准来要求并发瓶颈通常不在你的服务器而在大模型API的限流和响应时间上。认清这两点方案就很清晰了。我处理并发一共叠加了四层手段。第一层是流式输出SSE让用户不用等完整结果半天不吭声看到第一个字心里就踏实了。实现方式也很直接在FastAPI里返回StreamingResponse把Agent返回的token一块块推给前端。第二层是用户级限流用Redis令牌桶同一个用户每秒最多3个请求突发流量全部排队。第三层是结果缓存常见咨询问题按语义相似度命中缓存的话直接返回历史答案LLM只处理缓存未命中问题。第四层是连接池复用所有模型调用统一走异步HTTPX客户端避免每次请求都重新建连接。简单限流实现给大家看看import aioredis, time redis await aioredis.from_url(redis://localhost:6379/0) def check_rate_limit(user_id: str, max_requests: int 3, window: int 10) - bool: key frate:{user_id}:{int(time.time() // window)} count redis.incr(key) if count 1: redis.expire(key, window) return count max_requests实测数据10个并发用户同时进来没有限流时p95耗时在12秒左右加了缓存和连接池复用后降到6秒左右再上流式输出用户体感基本是无等待。如果要做批量型任务比如深夜晚群发、报表生成等耗时的操作建议再加一层任务队列Celery或RayHTTP层只负责接收任务并返回任务ID异步worker真去执行业务侧轮询结果这才是正解。注意不要指望通过高配服务器解决Agent并发问题。瓶颈在外部模型API的限流不是你的CPU。3.4 上线部署从本地到生产环境本地跑通只是第一步上线才是分水岭。我用了Docker容器化加nginx反代的组合。先给一份可用的多阶段构建DockerfileFROM python:3.11-slim AS builder WORKDIR /app COPY requirements.txt . RUN pip install --prefix/install -r requirements.txt FROM python:3.11-slim WORKDIR /app COPY --frombuilder /install /usr/local COPY . . EXPOSE 8000 CMD [gunicorn, -k, uvicorn.workers.UvicornWorker, -w, 4, -b, 0.0.0.0:8000, main:app]这里有个细节worker进程数不是越大越好。Agent接口本身是IO密集型异步模式下4到8个worker通常足够开太多反而因为上下文切换增加额外开销。nginx反向代理配置里最关键的是超时参数。默认的60秒代理超时会掐死LLM调用的长耗时请求必须调大server { listen 80; server_name agent.example.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_read_timeout 120s; proxy_send_timeout 120s; } }如果你的开发机是Windows或者虚拟机想模拟多站点环境可以直接在nginx里配多个server块分别绑定不同端口或不同自定义域名本地通过hosts文件做解析和线上结构完全对标。这个方法排查Agent回调里的跨域问题特别有用。部署流水线按照“构建镜像-推镜像仓库-服务器拉取-docker compose启动-健康检查-切流量”的顺序走。健康检查我额外做了一个独立的接口不完全等于FastAPI的/health而是要验证Redis和数据库连接都正常防止流量切上去才发现Agent依赖的后端挂了。4. 测试、监控与质量保障4.1 这个Agent该怎么测Agent系统测试比传统后端复杂很多因为它的输出有随机性。我分四层来测。第一层是单元测试把每个工具函数当普通Python函数测输入输出巨明确。第二层是图状态测试工欲善其事先验证某个输入下图的节点流转是否符合预期。第三层是场景回归测试预置一批典型问答对每次改代码后跑一遍对比回答的意图是否偏差。第四层是对抗测试专门投喂恶意输入、空输入、超长输入看系统崩不崩。一个最朴素的图测试代码长这样import pytest from agent.core import app # 编译好的agent图 def test_agent_ends_with_final_answer(): state {messages: [{role: user, content: 你好}]} result app.invoke(state) assert result[messages][-1].content ! assert !result[messages][-1].tool_calls这里提醒一个坑测试环境里的模型版本和生产环境若不一致场景测试基本等于白做。模型更新会导致输出风格和函数调用行为变化所以CI流水线里要锁定模型名称和版本号别让它悄悄漂移。4.2 监控与告警体系Agent上线后花最大的精力应该放在监控上。传统API的监控指标就那几项Agent系统还得额外盯六个关键指标指标含义告警阈值参考LLM调用次数/请求每请求平均调用模型多少次超过6次可能进入死循环token消耗/请求单请求总token量超过窗口警告p95响应时长端到端时延超过15秒预警工具调用成功率Agent调工具的失败情况低于90%预警上下文使用率每次请求上下文占用比例超过80%提示压缩缓存命中率语义缓存生效情况低于20%需优化日志要做到全链路可追踪每个请求生成一个request_id从入口到模型调用、工具调用每一关键节点都输出结构化日志字段至少包括request_id、node、action、latency、token数。上线第三周就是靠这套日志发现某个工具异常返回了超大JSON硬生生把上下文撑爆。否则这种问题靠肉眼排查根本找不到源头。5. 常见问题与排查技巧实录5.1 上线后最常踩的5个坑整理成速查表都是我这三次迭代里实际踩过并修复的现象根本原因解决办法跑几轮后回答质量变差上下文塞满工具输出和旧历史做消息裁剪摘要替换同一工具被反复调用工具返回结果模型不认识统一返回结构返回错误也明确写出来模型编造订单号工具描述里的参数边界不清晰描述里强调“参数必须来自用户输入”请求量高峰token成本猛涨缓存没生效所有问题都走模型加语义缓存同问同答首字响应太慢用户流失静态HTTP接口等全部结果才返回改SSE流式输出5.2 一个真实排查案例挑个印象最深的排查案例分享下。上线第二天运营反馈某个会话卡住一直没有回复。我打开结构化日志发现Agent在“agent”和“tools”两个节点之间来回跑了整整5轮每次都调用同一款工具而工具的返回结果都是同一串错误信息。原因很快浮出水面那个工具返回了一个非标准结构模型解析不了又不敢乱编只能硬着头皮重试同一个工具。修复方案是双管齐下在tools节点外包一层结果校验一旦发现工具返回非法结构直接把错误信息包装成“工具执行失败原因”返回给模型让模型换一种方式解决同时把该工具的描述改得更明确提示模型当工具失败时不要盲目重试。上线后同类问题不再出现。排查这类问题也有一些通用技巧分享给你们参考只要Agent表现异常先看日志里的节点链路和token趋势别急着改代码。给每个工具调用打上耗时标签超过预期耗时的调用重点看。模型输出和工具返回全部留存不要只存最终答案否则复盘没材料。复现问题优先缩小本地复现范围用固定输入和固定历史状态去跑图。5.3 后续可以这样扩展项目到目前为止已经能稳定支撑日常的查询和内容生成类请求。后面我还有三个扩展方向留给大家参考。第一是把长期记忆真正用起来接入向量库存储用户画像让Agent记住老用户偏好这个对电商和客服场景是刚需。第二是引入人工审批节点在涉及高危操作退款、发券、删数据时让Agent停下来等人工确认权限边界更清晰。第三是把Agent从单机切换到多租户模式每个租户一套独立的工具权限和提示词配置这是商业化SaaS化的基础。这次项目下来我的核心体会是Agent不是堆出来的Demo而是要靠工程约束把它驯服。模型会换、工具会变、业务需求也在迭代但“观察-反馈-修正”这条循环永远不变。只要你把观察的日志体系和反馈的测试基线打好这个Agent就能不断变强我自己在后面接手迭代时也是这么干的。