最近大半年我们团队一直在做一件事从零开始搭一套行业级的 AI Agent 基础设施把代码、配置、评测、部署全部沉淀成一个仓库。不是做一个能聊天的Demo也不是套一层大模型API就完事而是要把“智能体”当作一个正式的软件系统来对待。这篇文章就是我从0到1的实战复盘包括整体设计、仓库结构、核心模块、踩坑记录覆盖从代码到上线的全链路。如果你正准备在公司内部搭智能体或者想把自己手头的大模型应用提升一个级别这篇很值得看。1. 先想清楚行业级智能体到底在解决什么问题很多团队一上来就急着调大模型接口但聊了几轮之后就发现纯靠Prompt堆出来的“智能体”根本撑不住真实业务。所以我们做的第一件事不是写代码而是把问题定义清楚。1.1 智能体不是单纯的“大模型聊天框”大模型本身擅长的是“生成文本”它并没有手脚也不带记忆。给它一个需要多步操作的任务比如“查询订单状态并自动发起退款”它如果要完成就必须具备拆解任务、调用业务API、记录中间状态、处理异常等能力。把这些能力组装在一个“能自主行动的系统”上才是智能体。我常用的一个类比是大模型像一个非常聪明但没带工具和笔记本的顾问。你问他“明天天气怎么样”他能凭训练记忆给你个大概但如果你让他“查一下下一个小时会不会下雨再决定要不要安排户外会议”他就需要查天气API、读日历、判断用户偏好、最后给出建议。智能体就是给这个顾问配上“工具箱”“笔记本”和“工作流”。所以智能体 大模型 规划 工具调用 记忆 执行闭环。少了任何一个环节它都只是聊天机器人。1.2 Demo级和行业级的差距市面上有很多“智能体”Demo五分钟搭一个看起来很酷。但一放到生产环境就崩不记录状态、工具参数错乱、用户问一句“刚刚那个订单号是多少”就懵了、模型抽风时没人能干预。我们从头复盘了Demo级和行业级的核心差异整理成这张表维度Demo级智能体行业级智能体状态管理无状态每轮对话独立有会话状态支持多轮上下文恢复记忆靠上下文窗口硬塞分层记忆短期/长期可检索工具接入代码里写死一两个函数动态注册支持批量工具带参数校验可观测性print调试全链路日志、追踪、回放错误处理失败就重试经常死循环有降级、兜底、人工介入评测感觉“挺聪明”有评测集、指标回归、成本控制安全权限放开所有API最小权限、敏感操作二次确认性能不在乎有超时、限流、并发控制这张表列出的每一项都是我们在实战中踩过坑之后补上的。尤其是“评测”和“可观测性”它们决定了智能体维护三个月之后还是不是能掌控。1.3 我们的目标场景与选型思路我们选择的落地场景是企业内部知识问答 工单处理。用户会提出“帮我查一下报销政策”“这个订单为什么没发货帮我催一下”这类任务需要同时用到文档检索RAG、内部系统API调用、写数据库、发送通知。基于这个目标我们在技术选型上做了几个关键决策编排框架选择开源编排框架而不是自己手搓状态机。因为框架自带的并行、分支、状态管理、checkpoint能省掉很多底层的脏活。模型层以商用主流模型为主同时也保留了本地模型切换的接口。行业场景对输出稳定性要求高所以不是“哪个便宜用哪个”而是“哪个可控用哪个”。记忆层不直接依赖模型上下文窗口而是自建记忆服务把对话历史、业务事实、用户画像分开存储。工具层所有工具都走统一注册中心模型只能看到带有schema描述的工具而不是直接拼Prompt。选型的核心逻辑是降低试错成本保留后期替换能力。框架可以换、模型可以换但数据和评测体系不能换。下面我会按仓库的实际结构展开。2. 仓库结构一个可落地的Agent项目应该长什么样所谓“全仓库实战”就是整个项目不只包含“调用大模型的代码”还包括工具列表、评测用例、部署配置、监控脚本。我们的仓库挺庞大但核心目录结构很收敛。2.1 顶层目录设计我简化后的目录结构长这样agent-repo/ ├── agents/ │ ├── coordinator.py # 多智能体协调者 │ ├── planner.py # 任务规划器 │ ├── executor.py # 执行器 │ └── responder.py # 回复生成器 ├── core/ │ ├── memory.py # 记忆抽象接口 │ ├── memory_redis.py # Redis记忆实现 │ ├── context.py # 上下文组装 │ ├── tool_registry.py # 工具注册中心 │ └── config.py # 配置加载 ├── tools/ │ ├── internal_api.py # 内部业务API工具 │ ├── rag_search.py # 文档检索工具 │ ├── notify.py # 通知工具 │ └── db_ops.py # 数据库操作工具 ├── evaluator/ │ ├── cases/ # 评测用例YAML │ ├── metrics.py # 指标计算 │ └── run_eval.py # 评测入口 ├── deployment/ │ ├── Dockerfile │ ├── docker-compose.yml │ └── runtime_config.yaml └── scripts/ └── replay.py # 会话回放工具重点说几个容易被忽视的目录evaluator是评测集它是后期迭代的“安全网”scripts/replay.py是回放工具线上出问题后可以离线复现core/tool_registry.py是整个工具调用的中枢它是让Agent长出手脚的关键。2.2 核心模块之间的关系我们内部把Agent的运行理解成一个“感知-决策-行动-反思”的循环四个核心模块各司其职规划器Planner把用户请求拆成可执行的子任务。比如“查订单并催发货”会被拆成“查订单状态→查物流→判断是否超时→发送催货消息”。执行器Executor按照规划调用具体工具处理工具返回结果并把异常信息反馈给规划器。工具注册中心Tool Registry维护所有可调用函数的名称、描述、参数schema供模型选择和解析。记忆模块Memory保存当前会话的短期上下文、长期用户事实、业务对象状态。它决定了Agent能不能在多轮对话中“记得住事”。它们的数据流是用户输入 → 上下文组装 → 规划器决策 → 执行器调用工具 → 工具返回 → 更新记忆 → 生成回复 → 返回用户遇到复杂任务规划器会生成一个子任务列表执行器逐个执行执行结果会作为下一步决策的依据。整个过程是循环而不是单次调用。2.3 为什么需要多智能体协作我们一开始是“单Agent包打天下”但很快就遇到了问题文档检索逻辑和业务操作逻辑混在一起上下文特别长模型经常分不清该用哪个工具而且工具失败时整个对话就卡死了。后来我们改成了“多智能体协作”模式入口Agent负责理解用户意图检索Agent专门做RAG执行Agent负责调用业务API。这样做的两个好处职责隔离每个Agent的Prompt和工具列表都很精简模型决策准确率明显提升。风险隔离执行Agent出错时入口Agent还可以兜底不至于整个会话崩溃。但多智能体也不是越多越好。每多一个Agent就多一个出故障点。我们的建议是先用单Agent跑通再把稳定的子任务拆成独立Agent。3. 从0到1实现核心代码与关键配置这一节我们讲真正写进仓库的代码片段。我不能把整个仓库贴出来但会把最关键的骨架和设计思路分享出来你可以直接借鉴。3.1 配置加载与环境隔离行业级Agent一定不能把API密钥直接写进代码里。我们用Pydantic做配置模型从环境变量和配置文件读取参数# core/config.py from pydantic import BaseModel, Field from typing import Optional import os import yaml class AgentConfig(BaseModel): model_name: str Field(defaultgpt-4o, description主模型名称) temperature: float Field(default0.0, description行业场景尽量用低温度) max_iterations: int Field(default8, description单任务最大迭代步数) request_timeout: int Field(default30) tool_registry_path: str Field(default./tools/config.yaml) memory_backend: str Field(defaultredis, descriptionmemory or redis) redis_url: Optional[str] None api_key: Optional[str] None def load_config() - AgentConfig: path os.getenv(AGENT_CONFIG, ./config.yaml) with open(path, r) as f: data yaml.safe_load(f) # 环境变量覆盖配置 if not data.get(api_key): data[api_key] os.getenv(AGENT_API_KEY) return AgentConfig(**data)这里有两个细节值得注意temperature我设为0。行业场景里Agent的输出是要落库、触发操作的创造性越少越好。另外api_key不从yaml读默认值强制从环境变量注入避免误提交。3.2 实现一个可切换的记忆模块记忆模块是整个系统最重要的地基。我们分了两类记忆短期记忆当前会话的最近N轮对话直接放Redis带过期时间。长期记忆用户画像、历史偏好、业务实体状态放到向量数据库或者关系表里。接口设计得十分简洁# core/memory.py from abc import ABC, abstractmethod from typing import Optional, List class Memory(ABC): abstractmethod def add_history(self, session_id: str, role: str, content: str) - None: ... abstractmethod def get_recent_history(self, session_id: str, limit: int 10) - List[dict]: ... abstractmethod def save_fact(self, user_id: str, key: str, value: str) - None: ... abstractmethod def get_fact(self, user_id: str, key: str) - Optional[str]: ...Redis实现就很简单# core/memory_redis.py import json import redis from .memory import Memory class RedisMemory(Memory): def __init__(self, redis_url: str): self.client redis.from_url(redis_url) def add_history(self, session_id: str, role: str, content: str) - None: item json.dumps({role: role, content: content}, ensure_asciiFalse) key fagent:hist:{session_id} self.client.rpush(key, item) self.client.expire(key, 3600 * 24) self.client.ltrim(key, -20, -1) # 保留最近20条 def get_recent_history(self, session_id: str, limit: int 10) - List[dict]: key fagent:hist:{session_id} raw_list self.client.lrange(key, -limit, -1) return [json.loads(r) for r in raw_list]这里ltrim是防爆的关键。如果不限制短期记忆条数上下文会越滚越长成本也会失控。我们在具体业务里甚至把短期记忆再压缩成摘要在超长会话中只用摘要。3.3 工具注册中心让Agent能动起来工具注册中心负责三件事登记工具、暴露schema给模型、校验参数并执行。我们用装饰器模式实现# core/tool_registry.py from typing import Callable, Dict, Any import inspect import json class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict[str, Any]] {} def register(self, name: str, description: str, parameters: dict): def decorator(func: Callable): self._tools[name] { name: name, description: description, parameters: parameters, func: func, } return func return decorator def get_schema_list(self) - list[dict]: 返回给模型看的工具清单去掉函数对象 return [ { name: t[name], description: t[description], parameters: t[parameters], } for t in self._tools.values() ] def call(self, name: str, arguments: dict) - Any: tool self._tools.get(name) if not tool: raise ValueError(fUnknown tool: {name}) # 在这里可以插入参数校验、权限检查 return tool[func](**arguments) registry ToolRegistry()一个实际工具长这样# tools/internal_api.py from core.tool_registry import registry registry.register( namequery_order_status, description查询订单当前状态。当用户咨询订单是否发货、是否退款时使用。, parameters{ type: object, properties: { order_id: {type: string, description: 订单号例如 A10086} }, required: [order_id] } ) def query_order_status(order_id: str) - dict: # 模拟调用内部API return {order_id: order_id, status: 已发货, timestamp: 2025-06-01 12:00}给模型的schema越精确参数“幻觉”越少。我们曾因为描述里没写“订单号格式”模型生成了“订单ID”字段导致工具调用失败。所以每个字段都要写示例值描述要具体到格式和取值。3.4 规划器与执行循环规划器和执行器是整个循环的大脑与手脚。我们用一个轻量的主循环控制智能体不使用重型框架时它长这样# agents/executor.py class ReActLoop: def __init__(self, model, registry, memory, max_iterations8): self.model model self.registry registry self.memory memory self.max_iterations max_iterations def run(self, session_id: str, user_query: str) - str: history self.memory.get_recent_history(session_id, limit10) messages [{role: system, content: SYSTEM_PROMPT}] messages.extend(history) messages.append({role: user, content: user_query}) for step in range(self.max_iterations): response self.model.chat_completion( messagesmessages, toolsself.registry.get_schema_list(), tool_choiceauto, ) # 如果模型没有要调工具直接生成最终回复 if not response.tool_calls: final_answer response.content self.memory.add_history(session_id, assistant, final_answer) return final_answer # 否则执行工具调用 for call in response.tool_calls: obs self.registry.call(call.name, call.arguments) messages.append({ role: tool, tool_call_id: call.id, content: json.dumps(obs, ensure_asciiFalse) }) # 下一轮决策会将工具结果带上 raise TimeoutError(f超过最大迭代步数 {self.max_iterations})这段代码最重要的设计是tool_choiceauto让模型自己决定走“继续调工具”还是“结束并回复”。我们还在循环里加了最大迭代步数防止模型陷入死循环。实际生产中我们会把这个循环放进异步框架里并增加超时、重试、人工中断等控制。3.5 多智能体主从协作的连线前面提过我们拆了入口Agent和执行Agent。主从协作最简单的方式就是在入口Agent的工具列表里加一个“调用子Agent”的工具# tools/sub_agent_tool.py from core.tool_registry import registry from agents.executor import ReActLoop def create_sub_agent_tool(sub_agent_loop: ReActLoop): registry.register( nameinvoke_order_agent, description当用户需要查询订单、催发货、申请退款时调用订单处理子Agent。, parameters{ type: object, properties: { query: {type: string, description: 原始用户问题} }, required: [query] } ) def invoke_order_agent(query: str): sub_session_id fsub_{uuid.uuid4().hex} # 子会话ID result sub_agent_loop.run(sub_session_id, query) return {agent_result: result} return invoke_order_agent子Agent有自己的Prompt、工具列表和记忆空间。它跑完之后父Agent把结果当成普通工具返回值再组织语言回复用户。这样隔离了上下文也降低了相互干扰。不过要注意子Agent的调用是同步阻塞的如果遇到复杂任务最好用异步任务队列否则父Agent会等待很久。4. 工程化从能跑到可靠跑的跨越Demo能跑和行业级能用之间差了至少两万行逻辑。这一节我重点讲评测、可观测性、权限、部署这几个最容易被忽略的环节。4.1 评测集没有评测就没有“行业级”很多团队做Agent完全凭感觉“今天测试感觉不错上线吧。”结果一上线用户的问法稍微绕一点就崩。后来我们强制要求任何Agent改动必须跑评测集才能合并。我们的评测集全部来自真实历史工单和客服对话整理成YAML格式- id: case_001 user_query: 我上个星期买的东西现在还没发货订单号是 A123456能帮我催一下吗 expected_tools: - query_order_status - query_logistics - send_reminder expected_final_answer_contains: 已为您催单 tags: [order, reminder] - id: case_002 user_query: 报销差旅费需要什么材料 expected_tools: - rag_search expected_final_answer_contains: 发票 tags: [rag]评测指标我们不算“大模型评分”而是用偏工程的方式指标计算方式任务成功率正确完成全部步骤的用例数 / 总用例数工具调用准确率实际调用工具与期望工具的匹配比例参数合法性工具参数通过校验的比例平均轮数完成任务的平均LLM调用次数单任务成本每次任务的token费用这套评测集帮我们拦住了至少三四次回归有一次把检索Agent合并到入口导致所有RAG用例全部失败就是靠评测集发现的。4.2 可观测性日志、追踪与回放智能体是黑盒线上出问题如果没有日志调试会非常痛苦。我们把每次调用的完整轨迹记录下来用户输入每一步模型返回的推理内容选择的工具及参数工具返回结果最终回复token消耗、耗时日志是结构化的直接打到日志系统里{ event: agent_step, session_id: sess_001, step: 3, model_input_tokens: 2300, tool_call: { name: query_order_status, arguments: {order_id: A123456} }, tool_result: {status: 已发货}, ts: 2025-06-01T12:00:05Z }除了日志我们还写了scripts/replay.py可以读取历史轨迹逐行重放Agent的决策过程。出问题的时候把session_id给同学他几秒钟就能定位是第几步开始错的。这种能力对排查“模型幻觉”特别有效。4.3 权限与安全控制行业级智能体最大的安全隐患是“工具失控”。我们做过一次安全演练让普通用户发“帮我查一下所有用户的手机号”如果检索工具没有权限校验Agent真的会去执行。所以我们在工具层加了四道防线身份域隔离工具的Schema对人类是可见的但执行时会把用户身份透传进去业务API校验用户只能查自己的数据。参数白名单比如order_id必须匹配业务格式不允许传入SQL片段数据库工具只开放白名单方法。敏感操作二次确认写数据库、发短信、退款这类动作返回给用户“确认”按钮用户确认后才执行。可撤销设计Agent发起的通知都需要走消息中间件让运维能批量撤回。这里建议各位关注“权限下沉”而不是在Agent里做权限逻辑。Agent只负责“提出意图”最终权限判断由目标系统自己做否则迟早会被绕过。4.4 部署与灰度策略我们的Agent服务以异步Worker API网关方式部署。容器化打包存在镜像仓库里。每次发版采用“会话级灰度”新版本服务先接收10%的新会话观察成功率指标稳定后才全量。部署配置里最值得说的是限流与降级# deployment/runtime_config.yaml server: port: 8080 max_concurrent_requests: 50 request_timeout: 30s agent: fallback_strategy: rule_based # 当模型服务不可用时降级为规则引擎 max_retries: 2 tools: internal_api: { timeout: 5s, circuit_breaker: true } rag_search: { timeout: 3s, max_result_count: 5 }降级策略对我们的业务至关重要。如果大模型服务故障不能让用户完全得不到反馈。我们预先实现了一个“关键词规则引擎”模型挂了就按FAQ规则答复。虽然笨但能保证不被打爆。5. 常见问题与排查实录最后这一节全是真实踩过的坑。我把常见的几个问题按现象、原因、解决办法列出来希望你能少走弯路。5.1 模型陷入循环反复调用同一个工具却不结束现象Agent不断调用“查询订单状态”每次都返回同一个结果然后又问用户“要不要继续查看”来回绕。原因规划器没有从工具结果里提取关键信息也没有触发结束条件。模型不知道该何时停就会原地打转。解决设置max_iterations硬上限超过就返回“当前问题比较复杂请转人工”。在Prompt里设置明确结束条件“如果已经获取到用户问题的答案必须直接给出最终回复不要再次调用工具。”增加“重复动作检测”如果连续两轮的toolarguments完全一样中止循环。5.2 工具参数“幻觉”模型生成的参数不符合schema现象工具要的是order_id模型传了orderNumber工具要的是数字模型传了中文。原因一方面模型能力限制另一方面工具描述写得不够细。我们最初甚至出现过“order_id: 退货单号A12345”这种张冠李戴。解决工具描述中必须包含示例值“order_id: 字符串类型如A10086用户消息中一般以‘订单号’或‘单号’标识”。注册中心在调用前做严格校验失败后把校验错误返回给模型并附上“请根据错误信息修正参数”的提示让模型自行纠错。对关键工具开启“模型强制调用”tool_choice{name: query_order_status}保证先拿数据再说。5.3 上下文越拉越长成本暴涨现象多轮会话进行到第30轮时模型输入token已经超过5万一次调用成本好几块钱而且响应越来越慢。原因短期记忆没有上限且每次把全文都塞过去。模型还会把工具返回的大块JSON完整留在上下文里。解决给短期记忆设条数上限我们通常是10条超出的折叠成摘要。工具返回结果在进入上下文前先做精简只保留关键字段不把完整JSON塞进去。使用基于向量检索的长期记忆需要时再捞相关事实而不是全部加载。5.4 多智能体协作死锁父Agent等子Agent子Agent又调父Agent现象某个任务卡住日志显示父Agent调用了子Agent而子Agent又试图调用父Agent互相等待直到超时。原因工具注册表没有区分“哪个Agent可以用哪些工具”子Agent把父Agent的入口也当成了自己的工具。解决每个Agent实例配有独立的工具白名单子Agent只能看到分配给它的工具。所有跨Agent调用统一经过协调器协调器检测到循环调用时直接拒绝。调用链路上加全局唯一trace_id出现循环时能快速看到链条。5.5 常见问题速查表问题典型表现排查思路预防措施循环不收敛反复调用同一工具看回放日志检查是否有重复动作检测设置最大迭代步数、重复动作检测工具参数错误工具返回400/422检查schema描述、校验报错信息描述加示例、失败后重试上下文爆炸token数飙升、响应慢看日志中每步token消耗短期记忆截断、结果精简多Agent死锁任务卡住直到超时加trace_id追踪调用链Agent工具白名单、调用深度限制模型故障接口超时/返回乱码看模型服务状态、限流日志降级为规则引擎、自动重试权限泄漏用户查到自己无权数据权限测试用例工具层透传身份、目标系统校验这张表并不是终点。真正的经验是凡是出现过的问题都必须沉淀成自动化检查项。比如“循环不收敛”修复后我们立刻在评测集里加了对应的死循环用例之后每次发版都会跑防止复发。我个人在实际操作中的体会是搭智能体最难的其实不是算法而是工程。模型能力在快速提升但围绕模型做出来的记忆、工具、权限、评测、可观测性每一个都需要像搭传统后端一样认真对待。你越早把手感从“玩大模型”切换到“做系统”就越少踩后面的坑。最后再分享一个小技巧每次改动Agent时把旧对话回放一遍用肉眼看看新版本和旧版本的差异比什么指标都直观。