☰
从零搭建行业级AI Agent基础设施:架构设计与实战复盘
2026/10/18 10:29:53 网站建设 项目流程

最近大半年,我们团队一直在做一件事:从零开始搭一套行业级的 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的运行理解成一个“感知-决策-行动-反思”的循环,四个核心模块各司其职:

  1. 规划器(Planner):把用户请求拆成可执行的子任务。比如“查订单并催发货”会被拆成“查订单状态→查物流→判断是否超时→发送催货消息”。
  2. 执行器(Executor):按照规划调用具体工具,处理工具返回结果,并把异常信息反馈给规划器。
  3. 工具注册中心(Tool Registry):维护所有可调用函数的名称、描述、参数schema,供模型选择和解析。
  4. 记忆模块(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(default="gpt-4o", description="主模型名称") temperature: float = Field(default=0.0, description="行业场景尽量用低温度") max_iterations: int = Field(default=8, description="单任务最大迭代步数") request_timeout: int = Field(default=30) tool_registry_path: str = Field(default="./tools/config.yaml") memory_backend: str = Field(default="redis", description="memory 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_ascii=False) key = f"agent: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 = f"agent: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(f"Unknown tool: {name}") # 在这里可以插入参数校验、权限检查 return tool["func"](**arguments) registry = ToolRegistry()

一个实际工具长这样:

# tools/internal_api.py from core.tool_registry import registry @registry.register( name="query_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_iterations=8): 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, limit=10) 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( messages=messages, tools=self.registry.get_schema_list(), tool_choice="auto", ) # 如果模型没有要调工具,直接生成最终回复 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_ascii=False) }) # 下一轮决策会将工具结果带上 raise TimeoutError(f"超过最大迭代步数 {self.max_iterations}")

这段代码最重要的设计是tool_choice="auto",让模型自己决定走“继续调工具”还是“结束并回复”。我们还在循环里加了最大迭代步数,防止模型陷入死循环。实际生产中,我们会把这个循环放进异步框架里,并增加超时、重试、人工中断等控制。

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( name="invoke_order_agent", description="当用户需要查询订单、催发货、申请退款时,调用订单处理子Agent。", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "原始用户问题"} }, "required": ["query"] } ) def invoke_order_agent(query: str): sub_session_id = f"sub_{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真的会去执行。所以我们在工具层加了四道防线:

  1. 身份域隔离:工具的Schema对人类是可见的,但执行时会把用户身份透传进去,业务API校验用户只能查自己的数据。
  2. 参数白名单:比如order_id必须匹配业务格式,不允许传入SQL片段;数据库工具只开放白名单方法。
  3. 敏感操作二次确认:写数据库、发短信、退款这类动作,返回给用户“确认”按钮,用户确认后才执行。
  4. 可撤销设计: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里设置明确结束条件:“如果已经获取到用户问题的答案,必须直接给出最终回复,不要再次调用工具。”
  • 增加“重复动作检测”:如果连续两轮的tool+arguments完全一样,中止循环。

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时,把旧对话回放一遍,用肉眼看看新版本和旧版本的差异,比什么指标都直观。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询