资讯动态

生产级AI Agent开发实战:从工程架构到落地上线

发布时间:2026/10/20 13:20:40 来源:尧图企业网站定制
1. 动手之前想清楚你要做的 Agent 到底是哪一种市面上的 Agent 教程大多停留在“给大模型套一个工具调用循环”的层面跑通一个数学计算、查询天气的 Demo 就算完事。但真正放到生产环境里事情远没有这么简单并发一上来模型超时怎么办工具调用的结果格式不稳定怎么办用户连续问五轮之后上下文塞爆了怎么办这些问题在 Demo 阶段完全看不到上线之后却会逐一找上门。所以我想先泼一盆冷水生产级 AI Agent 和玩具 Demo 的分水岭不在模型强不强而在工程化够不够扎实。所谓生产级至少得满足三个条件——稳定不会因为一次工具异常就整体崩掉、可控成本和响应时间在预期范围内、可观测出了问题能快速定位到是模型决策错、工具返回错、还是代码逻辑错。在动手写第一行代码之前先回答下面 5 个问题这决定了你的整体架构长什么样。第一任务边界在哪里。Agent 不是万能的你要明确哪些请求必须交给 Agent 自主规划处理哪些请求走固定规则流程就好。我见过不少团队一开始就想做“全自动智能助手”结果需求发散得收不住最后连验收标准都定不下来。更务实的做法是从一个窄场景切入比如“售后工单自动分类与初步回复”先把闭环跑通再逐步扩展。第二记忆要做到什么程度。是单轮对话就够了还是需要跨会话记住用户偏好这直接决定你要不要上向量数据库、要不要做摘要压缩。很多时候“记忆”被过度设计实际上用一个简单的 Redis 缓存就能解决 80% 的场景。第三工具边界在哪里。Agent 能调用哪些工具、不能调用哪些工具每个工具的权限和操作范围有多大。这块如果不想清楚后面一定会出安全事故。我之前遇到一个团队给 Agent 接了数据库查询工具结果因为提示词写得含糊模型直接把整张表删了——虽然只是个测试库但吓得全组人出了一身冷汗。第四谁来为错误兜底。Agent 判断错了、工具执行坏了是让 Agent 自己尝试修正还是直接降级到人工处理还是抛一个明确错误给用户一定要提前设计好这个降级路径而不是指望大模型每次都能自己“悟”出来。第五成本预算的上限在哪里。每一轮对话背后都是 token 消耗加上工具调用的多次往返一次复杂任务的成本可能远超你的想象。要提前给单次任务设置 token 上限和费用上限超出就直接截断并告知用户。这 5 个问题想清楚之后再进入技术选型和编码阶段。我见过太多人跳过这一步直接开写结果写到一半发现架构撑不住需求推倒重来。磨刀不误砍柴工这句话放在 Agent 开发里再合适不过。2. 搭建基座工程结构、依赖与配置管理2.1 我的技术选型参考组合AI Agent 的技术栈没有标准答案但经过几个项目的对比我目前比较推荐这套组合它的思路适合大多数中小规模的生产级场景语言运行时Python生态最全所有主流框架都能直接接。API 服务层FastAPI异步支持好流式输出方便自带 OpenAPI 文档。编排框架LangGraph负责 Agent 的状态流转、工具调用、循环控制。它是 LangChain 的升级版把 Agent 建模成一张有向状态图比裸写 ReAct 循环要清晰得多。基础数据库PostgreSQL存用户、会话、工单这类结构化数据。缓存中间件Redis存短期会话记录、分布式锁、限流计数。向量库目前很多云数据库自带向量支持初期不用单独引一套 Milvus 之类的重组件用 PostgreSQL 的 pgvector 插件就够用了。为什么不用纯手写循环不是不能而是没必要。手写一个 ReAct 循环也就几十行代码但工程化之后你要自己处理状态持久化、并发安全、重试、超时、日志追踪这些都有成熟的坑LangGraph 把这套东西抽象成了图的节点和边你可以专注业务逻辑而不是重复造轮子。2.2 最小工程结构参考一个清晰的项目结构能帮你少走很多弯路。下面是我现在习惯用的目录布局你不需要完全照抄但建议保持层次一致agent-service/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置读取与校验 │ ├── api/ # 路由层 │ │ ├── routes.py │ │ └── schemas.py │ ├── agent/ │ │ ├── graph.py # LangGraph 状态图定义 │ │ ├── nodes/ # 各节点逻辑 │ │ │ ├── planner.py │ │ │ ├── executor.py │ │ │ └── memory.py │ │ ├── tools/ # 工具注册与实现 │ │ └── prompts/ # 提示词模板 │ ├── core/ │ │ ├── llm.py # 模型调用封装 │ │ ├── redis.py │ │ └── db.py │ └── observability/ # 日志、追踪、指标 ├── tests/ ├── docker/ ├── pyproject.toml └── .env.example这个结构的核心思想是分层解耦API 层只负责协议的进出Agent 编排层负责决策流转Tools 层负责与外部系统交互每一层都可以独立测试和替换。2.3 配置管理的几个关键点生产级应用的第一课就是配置不能写死在代码里。我把下面几类信息全部放到环境变量中模型 API Key、模型名称、Base URL。数据库连接串、Redis 地址。各工具服务的访问地址、密钥。运行环境标识devstagingprod。各类阈值参数超时时间、重试次数、token 上限。在代码里我用 Pydantic Settings 做配置加载和校验。它有一个我很喜欢的功能启动时自动检查必备环境变量是否缺失缺了就直接报错退出避免服务带着坏配置偷偷跑起来。from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, extraignore) app_name: str agent-service environment: str dev model_name: str gpt-4o-mini model_api_key: str model_base_url: str | None None database_url: str redis_url: str task_token_budget: int 20000 model_timeout_seconds: float 30.0 tool_timeout_seconds: float 15.0 max_agent_steps: int 8 settings Settings()注意这里我给每个超时和上限都设了默认值但核心密钥类配置没有默认值缺了会启动失败。安全底线不能有侥幸心理特别是生产环境。3. 核心引擎提示词、记忆、工具与主循环3.1 提示词分层把系统提示词当成代码来维护很多人在提示词上很随性直接在代码里写一大段字符串改一版就覆盖一版几个月后根本不知道线上跑的是哪段提示词。我建议把提示词当成一等公民来管理至少分三层第一层是系统提示词System Prompt定义 Agent 的身份、任务边界、行为准则。这一层要尽量稳定每次修改都要走 review 和测试。第二层是工具描述描述每个工具的功能、适用场景、参数含义和返回格式。这一层尤其重要因为模型决定调用哪个工具基本全靠这些描述。第三层是动态上下文包括用户输入、检索到的知识、历史对话摘要。系统提示词写作时有个经验直接告诉模型“能做什么还不够必须告诉它‘什么不能做’”。比如你给 Agent 接了数据库查询工具一定要在提示词里写清楚“只允许执行 SELECT 查询禁止任何 DMLDDL 操作违反此规则直接拒绝”。模型对边界的理解比对能力的理解更敏感负向约束往往比正向描述更能防止乱来。工具描述则要认真写清楚参数格式。模型比较擅长理解“传 JSON 对象”这种结构化描述但不太擅长理解“一个字符串格式为 xxx 格式字段用逗号分隔”。尽量把参数拆成结构化字段而不是拼接字符串这样工具调用的准确率会高不少。3.2 记忆系统短期会话和长期记忆分开存记忆是实现好 Agent 体验的关键也是最容易被过度设计的地方。我的实践方案是分两层短期记忆直接用 Redis 存最近 N 轮对话TTL 设为 24 小时。每次请求进来时把最近 20 轮对话拼进上下文。这样做的好处是不占数据库压力、自动过期、天然分布式友好。长期记忆就需要把对话做压缩摘要存入向量库。比如用户在第 3 轮提到“我喜欢简短回复”系统把这个偏好抽取成一条摘要存进向量库下次同一用户发起新会话时先用用户 ID 检索相关摘要把命中结果作为上下文注入。注意这里不是存原文而是存“对后续对话有价值的结论”。还有个细节容易被忽略记忆的写入时机。不要每一轮对话都往向量库写摘要那样会攒一大堆垃圾。我用的策略是当一个主题会话结束后用模型对整段对话做一次总结提取偏好和结论再写入向量库。简单说就是“会话结束才沉淀记忆”既能控成本又能控质量。不同记忆层的对比我整理了一张表记忆类型存储位置时效性用途成本短期会话Redis24 小时维持多轮对话连贯性低主要占上下文 token场景知识向量库长期提供业务领域知识中检索开销用户偏好向量库结构化字段长期个性化回复风格中写入成本高3.3 工具注册与调用统一协议收敛边界工具调用是 Agent 能力的放大器也是风险的集中地。我的做法是每个工具都写成独立的 Python 函数然后用装饰器注册到工具表中最后统一转成模型要求的格式。register_tool( namequery_orders, description查询指定用户的订单列表返回订单号、状态和金额, parameters{ user_id: {type: string, description: 用户唯一ID}, status: {type: string, enum: [pending, paid, shipped, completed], description: 订单状态可选}, }, required_permissionorder:read, ) def query_orders(user_id: str, status: str | None None) - dict: # 真实业务逻辑查库、拼接返回 ...这里有几个细节值得展开说。第一个是工具返回格式的标准化。所有工具必须返回统一结构状态码、业务数据、错误信息。这样 Agent 才能可靠地判断下一步动作而不是从一团乱字符串里猜结果。我习惯的返回结构是{code: 0, data: {...}, message: ok}非零 code 表示失败。第二个是 tools 描述字段的动态注入。如果一个工具依赖用户所在地区、会员等级这类上下文信息不要把它写死在描述里而应该在调用时动态传入。方法是先从上下文提取实参再拼到工具调用的 arguments 里。第三个是工具调用的安全校验。模型输出什么工具、什么参数你不加校验就直接执行这是生产事故的温床。至少在三个层面设卡权限校验这个用户有没有权限调该工具、参数校验参数类型、取值范围是否合法、操作校验危险操作二次确认。我之前做过一个客户案例模型被诱导调用了一个发送短信的工具参数里带着用户输入的电话号码差点造成短信轰炸。从那以后我所有工具的入口都加了一道参数校验和频率限制。3.4 主循环与状态机给 Agent 加上“刹车”纯 ReAct 循环最大的问题是没有“刹车”——模型如果在一个决策点上反复横跳它会无限循环下去烧掉大量 token用户还拿不到结果。解决这个问题最直接的办法是给循环加上三重限制最大迭代次数MAX_STEPS我常用 8 次超过就终止并返回“任务过于复杂请尝试拆分”。单轮 token 预算整个 Agent 的所有模型调用加起来不能超过设置的 token 上限。无进展检测如果模型连续两次输出几乎相同的决策说明它陷入了死循环直接切断。用 LangGraph 实现时Agent 的流程是一张显式的状态图。下面这段代码画出了我常用的核心流程from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated class AgentState(TypedDict): messages: list remaining_steps: int final_answer: str | None def decide_next(state: AgentState) - str: if state[final_answer]: return end if state[remaining_steps] 0: return max_steps last_message state[messages][-1] if last_message.get(tool_calls): return call_tool return respond graph StateGraph(AgentState) graph.add_node(plan, plan_node) graph.add_node(execute_tool, execute_tool_node) graph.add_node(respond, respond_node) graph.set_entry_point(plan) graph.add_conditional_edges( plan, decide_next, {call_tool: execute_tool, respond: respond, end: END, max_steps: END} ) graph.add_edge(execute_tool, plan)这段代码的价值在于把“循环控制”变成了显式的图结构。你可以很直观地看到每次工具执行完流程都会回到 plan 节点但每回到一次remaining_steps 就减一最终一定会走向终止。这就是生产级 Agent 和 Demo 的差别之一Demo 死循环了重启进程就行生产环境必须保证任何输入都会在有限时间和有限成本内返回结果。4. 从脚本到服务把 Agent 暴露成稳定可靠的 API4.1 流式输出SSE 的正确打开方式用户对着一个转圈圈等 20 秒才能看到完整回复这种体验不能忍。Agent 场景比普通对话更特殊因为中间可能穿插多次工具调用如果全程静默用户完全不知道系统在干什么。解决方式是用 SSEServer-Sent Events把过程事件实时推给前端。下面是 FastAPI 加 LangGraph 流式输出的核心写法from fastapi.responses import StreamingResponse app.post(/v1/chat) async def chat(req: ChatRequest): async def event_generator(): async for event in agent.astream({messages: build_messages(req)}): # 每个事件都推送一个 SSE 格式的 JSON yield fdata: {json.dumps(event, ensure_asciiFalse)}\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)前端需要注意两边一是事件流要分类型比如token表示增量文本、tool_call表示正在调用工具、done表示结束前端按类型渲染二是要处理 SSE 断线重连不要让用户一次次手动刷新。4.2 鉴权、限流与成本控制API 暴露之后第一件事不是写功能而是控制访问。生产级的做法是在网关层做 API Key 鉴权和 IP 限流应用层再做一层更细粒度的用户鉴权。限流不能只限请求次数对 Agent 场景更有效的是限 token 消耗。这个很容易理解和实现给每个用户设置一个每日预算每次请求结束后从额度里扣掉这次实际消耗的 token 数。扣到 0 就返回 429让用户下次再来。还有一点我想特别提醒一定要在调用模型之前估算 token 消耗而不是在模型返回之后才统计。否则并发高的时候预算会在某一秒被打穿。我用的方案是在每轮模型调用前先把消息列表转成 tiktoken 计数预扣一次预算调用结束后按实际用量多退少补。这套逻辑能保证成本不会失控但代码要细心预算的扣减和回补都要加分布式锁。4.3 超时、重试与降级策略Agent 服务最复杂的部分就是超时——它不是一个单一的模型调用而是“用户输入 → 规划 → 工具 → 再规划 → 响应”的完整链路链路里任何一环超时都可能中断整个任务。我的建议是设置三层超时形成一个完整的保护体系层级超时时间说明单次模型调用30 秒超过则重试一次再失败则降级单次工具调用15 秒短超时避免工具卡住拖死整个 Agent整条 Agent 链路120 秒链路超时后直接返回部分结果或错误重试策略也要分层设计。模型的 5xx 错误可以安全重试4xx 错误重试大概率也是同样的结果直接上抛更合理。工具调用要考虑幂等性——如果工具不支持幂等重试可能造成重复扣款或重复下单这时候正确的做法是“先查后写”先查一下上次的执行结果是否已经落库再决定要不要重试。降级路径是生产级 Agent 的最后一道防线。我常用的策略是A 方案跑完整 Agent 链路如果链路超时或模型不可用降级到 B 方案——直接返回一条固定提示“当前系统繁忙请稍后再试”对某些高优先级的场景可以再降级到 C 方案——转接人工客服。你必须在架构图上标好这条降级链路而不是在事故发生时现想应对方案。5. 上线前夜评估、追踪与可观测性5.1 没有测试集谈不上生产级很多人对 Agent 质量的评估停留在“我试了几个问题看起来不错”。这在 Demo 阶段没问题但生产级必须有量化的评估体系。我的做法是准备三类测试集黄金数据集Golden Set50 到 200 条高质量用户问题每条附带期望的工具调用序列和最终回答要点。这个集合人为编制并手动标注用于每次版本更新后的回归测试。场景测试集覆盖边界场景比如空输入、超长输入、无权限操作、工具执行异常、用户反复修改需求等。对抗样本集包含提示词注入、“忽略之前的指令”这类攻击性输入确保 Agent 不会越权执行危险操作。在评估方式上我比较推荐轨迹评估法和 LLM-as-Judge 结合。所谓轨迹评估是检查 Agent 每一步的决策是否正确比如应该调“查订单”工具的步骤有没有错误地调成“发消息”LLM-as-Judge 则是用一个更强大的模型对 Agent 的最终回答做 1 到 5 分的评分。两种方式结合既能发现过程问题也能评估最终质量。这里有一个很重要的细节评估要通过自动化的 CI 流水线跑。每次提示词变更、工具逻辑变更、模型升级都自动跑一遍测试集生成质量报告。只有报告通过代码才能合并。如果你现在还停留在“手动测试几个用例就上线”的阶段我强烈建议先补上这块。5.2 追踪每个决策Agent 的黑盒问题Agent 出错了最头疼的就是不知道问题出在哪一步。是模型理解错了是工具返回了脏数据是上下文被截断了没有追踪数据每次排查都是一场灾难。我推荐给 Agent 链路接 OpenTelemetry但要注意它的标准格式对“工具调用决策”这种事件表达力有限。更实用的做法是自定义一套追踪字段把这些信息单独记录下来追踪字段说明trace_id全链路唯一 ID用于关联所有日志step_index第几步规划model_input_summary该轮模型收到的主要上下文摘要llm_response模型原始输出包括 tool_callstool_name实际调用的工具名tool_args工具的完整入参tool_result_summary工具返回的摘要latency_ms每一步的耗时token_usage每一步的 token 消耗日志记录还有一个反模式要避开不要直接把完整 tool result 原封不动打到日志里。工具可能返回用户手机号、身份证号等敏感数据一旦日志平台泄密就是事故。我的做法是打“摘要 长度 关键字段脱敏”排查问题时按需去查明细。5.3 压测与性能基线上线之前必须知道系统的性能底线在哪里。我对 Agent 服务压测的关注指标和普通接口不一样吞吐量每秒能跑多少个完整 Agent 任务。由于任务是重计算吞吐量通常比普通 Web 服务低一两个数量级要有心理准备。P95 响应时间普通接口看 P99但 Agent 任务的响应时间受模型推理速度影响极大P99 会非常难看。我通常只看 P95 并控制在一个可接受的目标内。错误分布4xx、5xx、超时、模型限额各自的占比任何一个超过阈值都说明系统有问题。成本基线压测跑一小时统计 token 消耗总量。这个数字可以帮你估算生产环境一天的模型费用万一超出预算就需要提前做模型降级或任务简化。压测工具用常见的负载压测工具就行重要的是不要只压单一路径。要混合压一部分简单的“直接回答型”请求、一部分“需要调一次工具”的请求、一部分“复杂多步推理”的请求这才是真实流量的混合场景。6. 部署落地容器化、灰度发布与长期维护6.1 容器化的几个核心取舍把 Agent 服务容器化部署是标配但 Agent 场景有两个点和其他服务不太一样值得专门说一下。第一是“模型调用层”和“业务逻辑层”的分离。如果 Agent 业务逻辑层直接依赖单一厂商的模型 SDK后续切换模型或做模型路由时会非常被动。我在代码里加了一层 LLM Provider 抽象业务逻辑只面向抽象接口具体调 OpenAI、调开源模型还是调国内大模型由配置决定。这样切换模型不用改 Agent 核心逻辑只改配置。第二是镜像体积和启动速度。大模型的 Python 依赖普遍很重如果每个版本都重新打镜像、传镜像发布速度会很慢。我的做法是用多阶段构建构建阶段装完整依赖运行阶段只保留运行所需的最小文件并且提前把依赖缓存进镜像层。这样从代码提交到镜像就绪基本能控制在几分钟内。Dockerfile 的骨架可以这样写FROM python:3.11-slim AS base WORKDIR /app COPY pyproject.toml . RUN pip install --no-cache-dir . FROM base AS runtime COPY app ./app COPY prompts ./prompts EXPOSE 8000 CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]当然这是简化写法真实项目里你还需要处理健康检查探针、非 root 用户运行、TTY 关闭等安全加固细节。6.2 灰度发布与快速回滚Agent 服务的发布风险比普通 Web 服务高很多因为每次提示词调整都可能改变模型的输出行为而这种变化很难通过代码 review 发现。所以 Agent 服务的发布策略要格外保守。我的标准流程是新版本先部署到 staging 环境自动跑一遍黄金测试集通过后再切生产。生产的灰度分三步走先放 10% 流量观察错误率和用户反馈再放到 50%最后全量。灰度时间至少观察几个小时因为 Agent 的很多问题要在真实对话中才能暴露不是跑几个测试用例就能发现的。同时每个版本发布前都要保存上一个版本的“提示词 代码 配置”快照。如果灰度期间发现质量明显下滑直接一键切回旧版本。注意只有代码回滚是不够的提示词的回滚和模型版本的回滚同样重要。我之前踩过一次大坑代码回滚了但提示词是新的结果线上行为没有任何改善排查了半天才发现问题出在提示词上。6.3 上线后还会遇到的几个典型问题总结几个我在线上真实踩过的坑给准备上线的你打个预防针。第一个是上下文爆炸。Agent 多轮对话加工具调用后上下文会快速增长一旦超过窗口限制模型要么报错要么直接“失忆”。我现在会在每次规划之前检查消息列表的预估 token超过阈值就触发摘要压缩先让模型把前面的对话压缩成一段摘要替换掉历史消息。这个操作需要把压缩结果存到 Redis 里下次接着用。第二个是工具调用的幂等性问题。很多业务工具天然非幂等比如“发送工单通知”这种操作Agent 重试一次就可能重复发通知。我的解决方法是每个工具的执行都携带一个request_id工具内部根据这个 ID 做去重。这个 ID 由 Agent 主流程生成工具负责存起来并在收到重复 ID 时直接返回上一次的结果。第三个是并发下的状态隔离。Agent 的状态消息列表、步骤数、工具结果必须绑定到 trace_id 或者会话 ID 上不能放在全局变量里。我的项目初期就犯过这个错误线上两个用户同时提问结果 A 用户的工具返回结果被 B 用户的对话给覆盖了两个人都收到了一堆乱码。后来把所有状态都放进 LangGraph 的 State 对象由框架的线程隔离机制保证不串场问题才彻底解决。第四个是模型升级的隐性风险。厂商的模型接口保持兼容但具体的行为表现可能在某个时间点悄悄发生变化。之前我们有个 Agent 一直用得好好的某天突然开始频繁调用错误工具查了半天才发现是模型侧升级了。解决方法是锁死模型版本号或者加一层行为对比测试——每次例行升级模型都先跑一遍黄金测试集对比新旧版本的工具调用准确率有退化就立刻回退。最后再说一个经验生产级 Agent 的维护功夫主要在代码之外。定期把线上真实的 bad case 收集起来人工分析失败原因然后去补提示词、补工具校验、补测试集这个循环跑起来之后Agent 的质量才会稳步上升而不是每次发版都靠运气。我自己的习惯是每周固定抽半天做 bad case 复盘把典型问题写进评估测试集作为下一轮迭代的准入门槛。这套方法未必多高科技但长期坚持下来效果非常明显。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑