ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

AI Agent 工程化落地:从最小闭环到生产级稳定

AI Agent 工程化落地:从最小闭环到生产级稳定 AI Agent 开发是当前大模型应用中最容易“跑通一个 Demo、死在生产环境”的方向。很多人以为 Agent 只是把 Prompt 写得长一点再调用一次模型接口就能完成复杂任务但真正落地时会发现模型选型、工具注册、上下文管理、结果校验、成本控制和日志观测全都压在一起。如果不把这些环节拆清楚项目一旦接入真实用户和真实数据就会暴露大量不稳定因素。这篇文章以 AI Agent 应用开发为主线从最小可运行工程开始逐步讨论工具调用、上下文裁剪、模型部署、幻觉治理与线上排查。整篇文章面向有 Python 或 Java 基础、准备把 Agent 从实验demo 推向工程化项目的开发者不依赖某个固定大厂平台尽量保持与具体模型供应商无关。1. Agent 不是“长 Prompt”而是“完整执行链”1.1 模型调用只是第一步在常见项目中一个 Agent 至少包含四个组成部分模型负责理解用户目标并生成下一步动作通常是对话式大模型。工具负责执行模型无法直接完成的动作比如查数据库、调用业务接口、计算文件摘要。状态负责记录当前任务进行到哪一步、已经拿到哪些结果比如多轮会话历史。校验负责验证模型输出和工具返回是否正常避免把错误结果继续往后传。如果项目里只有一个模型调用代码那它更适合叫“聊天接口”不能叫 Agent。一个能工作的 Agent 必须循环执行“分析用户目标、决定调用哪个工具、执行工具、把结果回填给模型、继续分析”的完整链路。只有这个链路跑通并稳定才算真正落地。1.2 完整运行链路行动、观察、再行动拆开来看一次 Agent 任务通常按这个顺序进行接收用户输入。把输入和已有对话历史拼成上下文。模型返回回复内容或者在需要外部数据时返回工具调用请求。应用层解析工具调用请求执行对应函数。将工具返回结果作为新消息追加到对话中。再次调用模型让它基于工具结果生成最终回复。如果模型再次请求工具就重复第 4 到第 6 步直到模型返回自然语言结果。这个链路决定了三个关键工程要求工具执行结果必须能被模型理解不能只写日志。每一步调用都要考虑超时、失败和重试不能假设模型一次就能成功。上下文长度会不断增长必须有裁剪策略否则多轮任务一定会超出模型窗口。1.3 最容易踩的工程误区第一个误区是“工具函数直接写在 Prompt 里”。模型只能通过结构化工具描述知道有哪些能力具体执行必须走应用函数工具描述如果不准确模型就会频繁调用错误参数或编造不存在的工具。第二个误区是“一次调用就算完成”。用户问“帮我查一下北京天气然后生成一个穿衣建议”如果模型只返回一条天气文本没有真正调用天气工具这个结果可能是模型编造的。正确做法是强制模型先查工具再基于工具结果回答。第三个误区是“不处理失败结果”。工具返回超时、空数据、权限不足时模型可能继续基于假设编造。应用层必须把错误信息也回传给模型让模型知道这次查询不可用并主动向用户请求补充或放弃。2. 先搭建一个最小可运行的 Agent 工程2.1 技术栈选择Python 还是 Java目前 Agent 工程最活跃的生态在 Python 侧轻量方案可以直接使用 OpenAI 兼容接口加函数调用能力也可以使用 LangChain、LlamaIndex 等框架。如果所在团队以 Java 为主Spring AI 是更自然的接入点。这里给出一个选择维度不执着于“哪个框架最强”而是看团队维护成本技术栈适用场景主要成本典型工具链Python 原生 SDK快速验证 Agent 链路团队熟悉 Python需要自己实现工具路由、状态管理openai, pydantic, fastapiPython LangChain复杂 Agent、多工具、记忆组件多抽象层级多排查问题需要理解框架封装LangChain, langgraphJava Spring AIJava 服务内嵌入 Agent 能力统一技术栈Spring AI 演进较快版本需要锁定spring-ai, OpenAI 客户端接入公司内部模型网关生产环境统一走内部模型服务依赖网关能力和限流配额企业内网 SDK对于通读本文的读者建议先使用 Python 原生的 OpenAI 兼容接口跑通一个 Agent因为原生接口包装较少出现问题时能直接看到模型返回的 JSON 结构排查链路更短。2.2 项目目录结构一个适合学习的最小工程可以这样组织agent-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── agent.py # Agent 核心循环 │ ├── tools.py # 工具注册和实现 │ └── config.py # 配置读取 ├── requirements.txt ├── .env.example └── README.md这个结构足够小但已经区分了入口、Agent 循环、工具和配置。实际项目可以在此基础上增加 resources 目录、prompt 目录以及 test 目录但核心边界保持不变。2.3 配置外置不要把密钥写进代码在项目根目录创建.env.example内容如下OPENAI_API_KEYsk-xxx OPENAI_BASE_URLhttps://api.openai.com/v1 OPENAI_MODELgpt-4o-mini MAX_HISTORY_TOKENS4000 DEFAULT_TIMEOUT30加载配置时使用pydantic-settings或python-dotenv。生产环境不要直接依赖.env文件应由 K8s 环境变量或配置中心注入。这里的关键是代码里不要出现真实的 API Key仓库提交前要确认.env是否已经加入.gitignore。2.4 最小 Agent 代码下面代码实现了一个可运行的 Agent 循环支持带天气工具的例子。它只依赖 OpenAI SDK 和标准库方便看到完整执行过程。# app/tools.py import json from typing import Callable, Any _TOOLS: dict[str, Callable] {} def register(func: Callable) - Callable: _TOOLS[func.__name__] func return func def get_tool_schemas() - list[dict]: schemas [] for name, func in _TOOLS.items(): schemas.append({ type: function, function: { name: name, description: func.__doc__ or , parameters: { type: object, properties: { city: { type: string, description: 城市名例如北京, } }, required: [city], }, }, }) return schemas def run_tool(name: str, arguments: str) - str: func _TOOLS.get(name) if func is None: return json.dumps({error: funknown tool: {name}}, ensure_asciiFalse) try: args json.loads(arguments) if arguments else {} result func(**args) return json.dumps(result, ensure_asciiFalse) except Exception as exc: return json.dumps({error: str(exc)}, ensure_asciiFalse) register def get_weather(city: str) - dict: 查询指定城市的天气返回天气和温度。 # 示例工具只做演示。实际项目应调用真实天气服务。 if city 北京: return {city: city, weather: 晴, temperature: 23} return {city: city, weather: 未知, temperature: N/A}# app/agent.py import json import openai from . import tools from .config import settings def run_agent(user_input: str) - str: client openai.OpenAI( api_keysettings.openai_api_key, base_urlsettings.openai_base_url, timeoutsettings.default_timeout, ) messages [{role: user, content: user_input}] for _ in range(5): response client.chat.completions.create( modelsettings.openai_model, messagesmessages, toolstools.get_tool_schemas(), tool_choiceauto, ) message response.choices[0].message if message.tool_calls: messages.append({ role: assistant, tool_calls: [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in message.tool_calls ], }) for tc in message.tool_calls: tool_result tools.run_tool( tc.function.name, tc.function.arguments, ) messages.append({ role: tool, tool_call_id: tc.id, content: tool_result, }) continue return message.content or return 任务步骤超过循环上限已终止。说明几个关键点tools里通过register装饰器维护了一张工具表模型看到的 schema 和实际执行函数来自同一来源减少描述不一致。模型返回tool_calls时应用层需要把 assistant 消息和 tool 消息一起写回 messages。缺少任何一段模型就无法正确理解工具结果对应哪次请求。循环上限设置为 5防止模型反复调用工具导致请求失控。实际项目要根据任务复杂度调整并增加超时控制。3. 工具调用的工程化从“返回字符串”到“安全执行”3.1 工具函数的参数校验不能只靠模型自觉模型的 function calling 会输出 JSON 格式的参数但 JSON 结构正确不代表参数合理。比如模型可能把参数传成{city: null}或者传一个不在业务支持范围内的城市。应用层在调用工具前应该使用pydantic或jsonschema做一次显式校验。这里推荐用 pydantic 定义一个参数模型再让每个工具函数接收参数对象而不是直接解包字典from pydantic import BaseModel, field_validator class WeatherParams(BaseModel): city: str field_validator(city) classmethod def city_not_empty(cls, v: str) - str: if not v.strip(): raise ValueError(city 不能为空) return v.strip()在run_tool中先校验再调用def run_tool(name: str, arguments: str) - str: ... if name get_weather: params WeatherParams.parse_raw(arguments) result get_weather(**params.model_dump()) ...这样做有三个好处参数错误能在调用前暴露错误信息可以回传给模型工具函数内部逻辑更干净后续增加复杂参数时模型生成的参数可以被严格约束。3.2 多轮上下文回填是 Agent 稳定性的核心很多 Agent 在第一次工具调用后就开始乱答原因是 messages 回填顺序写错。消息顺序应该是user 消息 assistant 消息包含 tool_calls tool 消息每个 tool_call 对应一条 assistant 消息根据工具结果生成最终回复tool 消息的tool_call_id必须和 assistant 消息里的id对应。如果 id 对不上模型会混淆工具结果归属甚至直接报错。实际编码时不要手工生成 id必须使用模型返回的tc.id。3.3 上下文窗口裁剪不能无限追加消息一个真实 Agent 可能在一次对话里调用十几次工具每条工具结果可能都很长。如果不加控制请求 token 会不断膨胀最后超过模型上下文窗口请求直接失败。常见做法是按 token 估算而不是按消息条数裁剪。OpenAI SDK 没有内置 tokenizer可以使用tiktoken做 token 统计。简单实现如下import tiktoken encoding tiktoken.get_encoding(cl100k_base) def count_tokens(messages: list[dict]) - int: total 0 for message in messages: total len(encoding.encode(message.get(content) or )) tool_calls message.get(tool_calls) if tool_calls: for tc in tool_calls: total len(encoding.encode(tc[function][arguments])) return total在继续调用模型前先检查当前消息总 token 是否超过阈值。如果超过可以把最早的部分 user/assistant 消息折叠成摘要或者直接丢弃已经完成的工具调用中间过程只保留当前任务结论。这里要注意不要只按条数裁剪因为工具结果长短差异极大。一个数据库查询结果可能占 3000 token另一条普通回复可能只有 100 token。3.4 工具执行的安全边界必须写在代码里模型只能决定“调用哪个工具”不能决定“这个工具能做什么”。所以在应用层要设置工具权限边界高风险操作比如删除数据、修改用户资料、发送短信、执行 Shell 命令不要做成无条件自动执行。写操作建议先返回一个“待人工确认”的结果由用户或审批系统审核后再真正执行。工具函数内部要增加上限保护比如查询条数上限、文件大小上限、外部接口超时上限。对外部网络请求要设置 allowlist禁止模型任意指定 URL 去访问内网地址。实际项目里工具表最好和权限配置绑定。模型能看到哪些工具由当前用户角色和场景决定而不是把所有能力都暴露给模型。4. 模型部署与选型本地环境和生产环境要分开考虑4.1 模型选型维度模型选型不是“越贵越聪明”。需要结合任务复杂度、响应速度、token 成本和数据隐私一起评估任务类型推荐模型档位原因简单分类、关键词抽取小参数模型或远端轻量模型成本低、响应快多步 Agent 任务、复杂工具调用中高端模型指令遵循和 function calling 准确率更高涉及隐私数据的内部系统本地部署模型数据不出内网满足数据合规要求需要稳定低延迟的外呼、客服远程 API 缓存策略便于扩容和灰度4.2 本地部署与远程 API 的取舍本地部署常见工具是Ollama、vLLM或llama.cpp。远程 API 可以直接使用 OpenAI 兼容协议。两者不是互斥关系很多生产架构会做成模型网关默认走远程 API敏感场景路由到本地模型。对比项远程 API本地部署部署成本低开通即可高需要 GPU 资源和运维数据链路数据发送到外部服务数据只在内网延迟受网络和排队影响取决于 GPU 推理速度可扩展性供应商负责扩容自己处理并发和负载模型版本由供应商控制可固定版本并自测费用按 token 计费固定硬件成本学习环境建议优先用远程 API 跑通功能不要一上来就买显卡部署模型因为 Agent 的稳定性和模型部署没有直接关系先解决业务链路再根据隐私和成本要求决定部署方式。4.3 请求参数超时、重试、并发和限流Agent 的请求参数不宜直接使用 SDK 默认值。实际生产环境至少要关注以下参数参数默认值参考调小影响调大影响timeout30s快速失败但可能误杀慢推理避免误判但请求积压会拖垮资源max_retries0 或 1失败直接返回业务不友好提高成功率但可能放大重复请求max_tokens根据场景设置回复变短长任务被截断增加 token 成本和等待时间temperature0.2输出更稳定但可能重复输出更多样但幻觉风险升高对于 Agent 场景推荐temperature设置为 0.2 以下因为工具调用需要确定性不需要创造性。工具调用结果如果不能稳定复现测试和排错都会很难做。并发和限流也要在应用层实现。模型 API 供应商通常有 RPM每分钟请求数和 TPM每分钟 token 数配额如果 Agent 内部循环多次调用模型一次用户请求就可能消耗多次配额。生产环境要用令牌桶或信号量做限流避免瞬间打爆配额。import asyncio from semaphore import Semaphore agent_semaphore asyncio.Semaphore(10) async def call_agent_with_limit(task): async with agent_semaphore: return await task这里只是示例。实际上要结合业务场景设置并发上限并监控每次请求的耗时和 token 用量。4.4 生产环境还要加模型网关和缓存不是每次对话都需要调用大模型。固定的工具返回结果可以使用缓存比如高频查询“北京天气”如果模型已经调用过工具结果可以在短时间缓存。模型回复也可以加语义缓存但要谨慎避免把用户隐私写入缓存。更稳妥的方式是在应用和大模型之间加一个内部模型网关。网关负责多供应商路由和降级token 统计和成本分摊敏感信息过滤请求日志和采样配额管理和告警网关的实现本身是一个独立服务可以使用组件也可以自研。对大多数中大型项目这一步是必要的否则后续扩容和成本治理会很困难。5. 幻觉治理与测试不能只看“返回有没有内容”5.1 幻觉从哪来Agent 场景下幻觉不只是“模型编造事实”还表现为工具调用生成错误参数导致业务数据查询失败。工具结果为空时模型仍然基于自己的知识补全答案。模型把一次调用中看到的工具结果错误套用到另一个问题。模型在上下文被截断后丢失前期结论继续生成看似完整但逻辑断裂的内容。排查幻觉的第一步不是让模型“不要胡编”而是确认模型拿到的输入是否完整、工具结果是否真实、上下文是否被错误裁剪。5.2 用测试集量化效果建议为 Agent 建立一组回归测试用例每个用例包含{ id: case_001, query: 北京今天需要带伞吗, expected_tool: get_weather, expected_tool_args: {city: 北京}, expected_no_unsupported_fact: true }测试分两层工具调用层断言模型是否调用了正确工具参数是否满足 schema。最终回复层断言回复中没有编造工具结果格式符合预期。工具调用层比最终回复层更容易自动化。可以先只做工具层断言再逐步增加回复校验。5.3 用结构化输出约束答案对于需要返回给其他系统处理的 Agent推荐用 JSON Schema 约束模型输出。比如要求最终结果必须是{ city: 北京, weather: 晴, temperature: 23, advice: 不需要带伞 }在模型调用中通过response_format或 structured output 参数强制 JSON 输出再用 pydantic 校验。这样即使模型回答内容不完美至少结构稳定下游系统不会解析失败。注意在这里不要用正则或者字符串查找去解析模型答案。大模型输出格式稍有变化就会让解析失效结构化输出才是更稳的方案。5.4 可观测性每次调用都要能回放生产环境出现“用户说回答不对”时排查者最需要的信息是完整请求回放。每个 Agent 请求至少记录request_id 和 session_id每一轮模型请求的 messages模型返回的 tool_calls 和最终回复每一步的耗时和 token 用量工具的入参和返回结果是否命中了上下文裁剪日志建议使用 JSON 格式方便接入统一日志平台。采样率不一定是 100%但对于高风险用户或异常任务必须全量记录。6. 常见问题排查现象、原因、处理方式6.1 工具从未被调用现象模型始终返回自然语言不调用工具。可能原因工具 schema 里description太模糊模型不知道何时使用。tool_choice设置成了none。模型版本较老不支持 function calling。消息格式里tool_calls字段名错误。检查方式打印模型原始返回确认finish_reason是否为tool_calls。然后检查工具 schema 是否在请求中正确传递。解决建议为工具描述补充触发条件和示例比如“当用户询问天气时先调用 get_weather 获取天气不要直接回答”。6.2 工具被调用但参数一直错误现象模型生成的参数缺字段、多字段或类型错误。原因多为 schema 不够严格。比如没有把必填字段写进required没有给属性加清晰的description。处理建议使用框架自动生成 schema而不是手写两份 JSON。随后增加 pydantic 参数校验校验失败信息回传给模型让模型看到具体的参数错误。6.3 循环多次仍不结束现象Agent 在工具调用里转圈最终触发循环上限。原因通常是工具结果没有真正帮助模型完成用户目标或者上下文里没有提示“如果已经完成请直接回答”。处理建议是增加循环上限并在系统 Prompt 中写清楚结束条件。6.4 请求报上下文长度超限现象请求直接 400提示maximum context length。原因就是 messages 累积过多。需要统计 token 并提前裁剪。重要工具结果可以摘要化不要全量回填。6.5 工具执行成功但最终回答明显错误现象工具返回了天气晴模型却回答会下雨。原因可能是模型在上下文压力下忽略了工具结果过度依赖自己的内部知识。此时应增强系统提示词要求“只能根据工具返回内容回答”同时对最终答案做规则校验但不要期望 100% 消灭。下表汇总排查路径问题现象常见原因检查方式处理建议工具从未被调用schema 描述不清晰或 tool_choice 错误打印模型返回 finish_reason补充触发条件和示例设置 tool_choice参数总是错误schema 缺少 required 和类型约束检查模型传参 JSON用 pydantic 校验并把错误回传多次循环不结束没有结束条件或工具结果不够查看循环日志增加循环上限系统提示词写明结束条件上下文超限messages 不断膨胀统计 token 分布写摘要或裁剪早期消息回复与工具结果矛盾模型忽略工具结果对比工具结果和最终回复强化只依据工具回答的约束7. 让 Agent 更稳定的几条工程建议7.1 先跑通“最小闭环”再增加框架不要一上来就引入 LangChain 或 Spring AI。先用原生 SDK 把工具调用跑通理解 messages 回填规则再考虑是否需要框架。框架能加速开发但也会隐藏细节排查问题时反而增加认知负担。7.2 把提示词当作代码来管理Prompt 不能只写在字符串里后续要改配置。建议把系统提示词抽离到独立文件保存版本并记录每版本对应的测试结果。生产环境出现行为变化时能快速定位是代码变更还是提示词变更导致。7.3 为每个工具设置失败语义工具返回不能只返回数据还要包含状态字段。比如{ status: success, data: {city: 北京, weather: 晴} }失败时返回{ status: error, error: 天气服务超时 }这样模型能区分“没有数据”和“数据正常但结果就是这样”避免模型把失败场景当成正常结果继续编造。7.4 灰度发布和回滚策略Agent 依赖大模型服务模型供应商可能升级版本导致行为变化。项目应保留模型版本参数通过配置中心切换。每次上线先灰度 5% 流量观察工具调用成功率、平均耗时、回答格式正确率和用户反馈再逐步放开。7.5 学习环境与生产环境分离学习环境可以只用一个 API Key 和本地代码。生产环境则至少需要模型网关、日志平台、监控告警、权限系统和配额管理。不要因为 Agent demo 跑得通就直接在生产开放权限尤其要避免把后台管理工具直接暴露给 Agent 自动调用。7.6 从一个小工具开始练习对新手来说最好的练习不是写一个“万能助理”而是做一个只包含两个工具的 Agent比如“查天气”和“查日历”。先让模型能稳定选择工具、执行工具、回填结果再慢慢增加工具数量和上下文复杂度。每一步加一个变量出问题时能更快定位。AI Agent 的工程难点不在于模型本身而在于模型、工具、状态、校验和基础设施如何稳定衔接。把最小链路跑通把工具调用参数约束好把失败结果回传清楚把上下文裁剪策略想好再把日志和监控补上这个 Agent 才具备进入生产环境的资格。
返回列表