☰
Agent Infra实战:从零搭建企业级Agent基础设施的关键技术
2026/10/13 4:20:53 网站建设 项目流程

1. 为什么Agent Infra突然成了“隐秘战场”

这两年大模型本身的热度已经在肉眼可见地下滑,圈内人都在说“模型层卷到头了”,真正还在闷声布局、抢人抢资源的地方,其实是Agent的基础设施层,也就是Agent Infra。这个领域不太有声量,因为面向开发者而不是大众消费者,但巨头们在这个赛道的布局密度,远比我预想的高得多。

我在某公司做后端架构的工作,去年开始陆续接触Agent相关项目。刚开始我和大多数人一样,觉得Agent不就是调几个API、套个Prompt模板、让模型走几步工具调用嘛,实在没什么技术含量。直到我们自己做了一个跨平台系统的Agent化改造,我才真正意识到,Agent做到真正稳定、可控、可运维、可观测的时候,就变成了一套全新的基础设施工程,而这套工程里面每块都是硬骨头。

巨头们在争什么?表面上是抢大模型API的调用入口,实际上是在抢Agent运行时的控制权。谁掌握了Agent运行时的编排、记忆流转、工具调用链路,谁就掌握了未来AI应用中最高价值的那一层抽象。你想想看,大模型本身更像一个发动机,而Agent Infra是底盘、刹车、仪表盘和悬挂系统,发动机再好,没有底盘也跑不起来。真正开过车的人都知道,底盘调校的复杂度远远超过发动机本身。

对普通工程师来说,这反而是个好机会。巨头和巨头之间的竞争,实际上把整个赛道的工具链、开源方案、行业标准都催熟了,我们不用自己踩一遍所有的坑,不少基础设施级的方案已经有开源实现。作为工程师,入场的方式很简单:把手头现有的业务系统往Agent方向改造一遍,理解其中哪一块儿最痛、最缺基建,那就是我们的入场券。

2. Agent Infra到底包含哪些内容

先说结论:Agent Infra不是某一个具体组件,而是围绕“Agent的运行、记忆、工具、评估、安全”建立起来的一整套支撑体系。我把它拆成五个核心面,方便大家理解。

2.1 运行时与执行引擎

Agent运行时是整个体系的骨架。它负责调度大模型调用、工具调用、用户指令之间的循环,核心要解决的是“让Agent在多次推理与动作之间不迷路”。

我们用的方案是事件驱动架构。模型每推理完一轮,会输出一个意图,然后运行时把这个意图解析成可执行的任务步骤,再触发对应的工具服务。这套设计的核心难点在状态管理:Agent执行一个有五步甚至十步的任务时,中途哪一步挂了、哪一步超时了、哪一步返回的数据格式不符合预期,运行时都要能感知到并做出补偿。

实际做下来,最崩溃的就是时序问题。模型生成工具调用参数是毫秒级的事情,但工具真正返回结果可能需要几秒甚至更久。如果运行时只是简单同步等待,整个Agent就卡死了。我们最后用异步事件流来解决问题,模型先把调用意图丢进队列,工具服务执行完把结果重新注入上下文,运行时再决定下一步动作。这套机制本身不复杂,但稳定性要求极高,消息丢失、重复投递、结果乱序都可能导致Agent整个推理链条乱掉。

2.2 记忆与状态管理

记忆是Agent区别于普通对话机器人的核心能力。但记忆在工程上的实现,远比想象中坑多。

我们一开始天真地用Redis把所有对话记录存下来,每次请求直接塞进Prompt里,后来发现上下文塞爆是迟早的事,而且塞进去的内容大部分是无效噪音,模型反而不容易抓住重点。后来我们把记忆拆成短期记忆和长期记忆:短期记忆是当前任务内的会话上下文,长期记忆则按用户维度存储,并增加一个摘要和检索的过程。

长期记忆的落地方式参考了向量检索的思路,先把历史交互记录做向量化,然后根据当前任务需求做语义检索,选出最相关的几条作为参考上下文。这听起来很顺理成章,但实际操作中,记忆写入的触发时机、摘要的更新策略、记忆的过期清理,全都要手写代码控制。很多Agent项目之所以越跑越“笨”,就是因为记忆里堆了大量过时、矛盾甚至错误的信息,模型再怎么聪明也会被带偏。

2.3 上下文工程

上下文工程是我个人认为Agent Infra里最微妙的一块。它不是简单地把背景资料塞给模型,而是要在“给多少信息、给什么粒度、以什么顺序给”之间做取舍。

我们内部封装了一个上下文管理服务,每次模型推理前,动态组装机器学习到的上下文:用户画像、历史记忆摘要、当前任务状态、工具调用结果、系统提示词,每一部分都有单独的优先级和拼接策略。关键的经验是:上下文必须在任务推进中持续压缩和重写,而不是一次性固定下来。比如一个多轮工具调用的任务,每一步执行完之后,原始返回值不一定还需要原样放在上下文里,这时候可以做一个结构化抽取,把关键字段提炼出来,替换掉冗长的原始数据。

这块没有标准答案,完全靠实验和经验积累。但有一点是确定的:上下文设计决定了Agent能力的上限。模型本身的能力再强,上下文给糊了,照样输出垃圾内容。

2.4 可观测性与评估体系

Agent系统的调试难度,比传统微服务高一个数量级,核心原因是不确定性。同样的输入,模型每次的输出都不完全一样,工具调用链条也可能走不同分支,所以传统按日志关键字检索定位问题的方式几乎失灵。

我们的做法是引入Tracing机制,为每一次Agent执行生成唯一的Trace ID,贯穿从接收指令到模型推理到工具调用再到最终回复的全链路。每一步都记录入参、出参、耗时、token消耗,并且额外记录了模型当时选择的“思路过程”,这样才能在出问题的时候还原决策路径。

评估体系是更头疼的事情。普通接口测试是断言“返回的JSON是否包含某某字段”,Agent评估没法这么干,你得判断“这个回答是否完成用户意图”“工具调用步骤是否合理”“有没有产生幻觉信息”。我们最后做了三层评估:规则层拦截明显失误,语义层用大模型打分,人工抽检层定期复盘。是的,你没看错,用大模型评估大模型,这是当前没有人能完全绕开的路子,但必须作为兜底而不是唯一依赖。

2.5 安全与护栏体系

Agent比传统API多了“行动力”,所以安全要求完全不同。传统接口最多是数据泄露,Agent可以调用工具,可能触发删除操作、写数据、发消息等高危行为。

我们的做法是把工具调用全部纳入审批和流控:高风险操作必须二次确认,低风险操作也要走频控和配额限制。每次工具调用前,会有一个独立的校验器,对参数做白名单检查和类型校验,避免模型生成恶意或者错误参数直接打到下游系统。另外,Prompt注入是Agent特有的大坑,用户可能故意在输入里塞“忽略前面的指令”,这时候不能光靠模型自觉,还得靠输入侧过滤和指令权限隔离来兜底。

3. 巨头之间的竞争逻辑,和我们工程师有什么关系

巨头在Agent Infra的布局,看着离我们很远,实际上直接影响着我们手里的技术选型和职业方向。我理解他们的布局主要有三层逻辑。

第一层是模型入口之争。大模型本身越来越像标准化商品,API价格一路走低,纯靠模型赚钱越来越难。真正有黏性的是Agent平台:开发者在平台上配置工具、构建Agent、沉淀工作流,用习惯了就很难迁移。这就像早年间手机厂商抢生态入口一样,最终锁定的是开发者习惯和生态壁垒。

第二层是数据回流。Agent运行过程中产生海量的调用链数据、用户偏好数据、任务成功失败数据,这些数据对模型迭代和产品优化价值巨大。基础设施层做得越好,越能收集到高质量的数据飞轮。

第三层是标准制定权。开源框架的接口规范、工具调用协议、记忆存储格式,谁参与得多、贡献得多,谁的话语权就大。毕竟大家都不希望自己的技术栈绑在某一家厂商上,但也不希望自己在标准制定中缺席,这就是很多巨头一边推自家闭源产品,一边积极往开源项目里提交代码的原因。

工程师能从中得到什么?两条非常现实的红利:

一是工具链红利。巨头打架,开源项目受益,我们内部很多方案现在不需要从零手写了。生态越成熟,Agent开发门槛越低,懂底层原理的人就越值钱。

二是差异化红利。巨头做通用基础设施,但落地到具体行业、具体业务,还是需要有人懂业务、懂系统集成。这就是工程师的机会:把通用Agent能力翻译成某个垂直领域里真正可用的方案。通用基础设施和垂直落地之间,存在一个巨大的空白层,属于两头都不太愿意深度扎进去的地方,但它恰恰是工程师施展空间最大的区域。

4. 实操过程:从零搭一个最小可用的Agent运行框架

理论说再多,不如直接上手跑一套。下面是我们内部做的第一个Agent基础设施原型的过程,不算多高级,但足够你理解这套体系的运行逻辑。

4.1 整体架构设计

我们规划了一个极简的Agent Infra骨架:

  • Agent运行时:负责任务拆解、状态流转、模型调用编排
  • 工具注册中心:维护所有可供Agent调用的外部能力接口
  • 记忆服务:管理短期记忆和长期记忆的读写与检索
  • 评估服务:对Agent每次执行结果做自动评估
  • Trace追踪:记录完整调用链

技术栈选择上,我们用了一个比较成熟的异步任务队列框架做事件流转,模型层接了多种大模型API,统一走一套协议适配层,这样切换不同的模型提供方时,不会改动上层编排逻辑。

4.2 核心代码骨架

运行时最关键的是Event Loop机制,直接决定Agent的并发处理能力。我把核心骨架简化了一下。

# runtime_loop.py import asyncio from enum import Enum class AgentState(Enum): IDLE = 0 PARSING = 1 PLANNING = 2 EXECUTING = 3 FINISHED = 4 FAILED = 5 class AgentRuntime: def __init__(self, memory_client, tool_registry): self.state = AgentState.IDLE self.memory = memory_client self.tools = tool_registry self.event_queue = asyncio.Queue() self.trace_id = None async def submit_task(self, user_input, trace_id): self.trace_id = trace_id await self.event_queue.put(("parse", user_input)) async def start(self): while True: event_type, payload = await self.event_queue.get() if event_type == "parse": await self.handle_parse(payload) elif event_type == "plan": await self.handle_plan(payload) elif event_type == "execute": await self.handle_execute(payload) elif event_type == "finalize": await self.handle_finalize(payload) break async def handle_parse(self, user_input): # 调用模型解析用户意图 intent = await self.model_parse(user_input) await self.event_queue.put(("plan", intent)) self.state = AgentState.PLANNING async def handle_plan(self, intent): # 根据意图决定是否需要调用工具 if self.tools.is_tool_call_required(intent): tool_name, params = await self.model_decide_tool(intent) await self.event_queue.put(("execute", (tool_name, params))) else: await self.event_queue.put(("finalize", intent)) async def handle_execute(self, tool_task): tool_name, params = tool_task # 异步执行工具调用 result = await self.tools.invoke(tool_name, params) # 将结果写回记忆上下文 await self.memory.update_current_context(self.trace_id, tool_name, result) await self.event_queue.put(("finalize", result)) async def handle_finalize(self, final_payload): # 生成最终回复 answer = await self.model_generate(final_payload) self.state = AgentState.FINISHED return answer

这套骨架逻辑上很直观,但实际生产化要补的东西非常多。每个异步任务要做超时控制,队列里的事件要做优先级区分,工具调用要做并发限制,模型解析失败要有重试和降级策略。我们第一次上线时没考虑工具调用超时,结果某个外部接口响应卡了几分钟,直接把整个事件循环堵死,所有Agent全部挂起。后来才加上超时和熔断逻辑:

# tool_invocation.py import asyncio import functools async def invoke_with_timeout(tool_func, timeout_seconds=10): try: return await asyncio.wait_for( tool_func(), timeout=timeout_seconds ) except asyncio.TimeoutError: # 返回一个明确的超时标记,而不是抛出异常 return { "status": "timeout", "partial_result": None, "retryable": True }

4.3 记忆模块的实现要点

记忆模块的细节决定Agent的“智商持续在线程度”。我们用了两层存储:短期记忆放Redis,长期记忆放向量数据库。

短期记忆以Trace ID为Key,保存当前任务所有的交互记录、中间状态和工具返回结果。长期记忆只在任务结束时进行一次沉淀:把本次任务的摘要、关键决策、用户偏好抽出来,向量化后写入。

# memory.py class AgentMemory: def __init__(self, redis_client, vector_client, summarizer): self.redis = redis_client self.vector = vector_client self.summarizer = summarizer async def save_short_term(self, trace_id, role, content): await self.redis.rpush(f"agent:{trace_id}:ctx", f"{role}:{content}") async def load_relevant_long_term(self, user_id, query, top_k=5): query_vector = self.vector.embed(query) results = await self.vector.search( index="agent_memory", query_vector=query_vector, top_k=top_k, filter={"user_id": user_id} ) return results async def finalize_memory(self, trace_id, user_id): # 拉取整条短期记忆 raw_ctx = await self.redis.lrange(f"agent:{trace_id}:ctx", 0, -1) # 调用摘要模型提炼长期记忆 summary = await self.summarizer.summarize(raw_ctx) await self.vector.upsert( index="agent_memory", vectors=[{ "id": trace_id, "vector": self.vector.embed(summary), "metadata": { "user_id": user_id, "summary": summary, "timestamp": datetime.now().isoformat() } }] ) # 清理短期记忆 await self.redis.delete(f"agent:{trace_id}:ctx")

实操下来最大的反馈是:长期记忆的写入时机非常关键,不是所有任务都值得沉淀。如果一笔交易只查了一个订单状态,这也要写入长期记忆,用不了多久向量库里全是噪音。我们后来加了业务规则过滤,只有模型判断对用户有长期价值的交互才触发沉淀。

4.4 工具调用协议设计

工具调用的协议设计,同样是Agent Infra的核心环节。大模型没有“天然知道你的系统接口怎么调用”的能力,需要我们把工具能力描述清楚。我们为每个工具维护了一份JSON Schema格式的声明,包括功能描述、参数类型、枚举值、示例和必要的调用约束。

{ "tool_name": "query_order_status", "description": "根据订单号查询订单当前状态和物流信息", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "订单号,通常为字母和数字组合" }, "include_logistics": { "type": "boolean", "description": "是否返回物流轨迹详情", "default": false } }, "required": ["order_id"] }, "constraints": { "rate_limit": 10, "timeout_ms": 3000, "requires_confirmation": false } }

这个描述文件的质量,直接决定模型工具选择的准确率。早期我们写着“查询订单状态”五个字就完事,模型经常把物流查询和订单查询混在一起,参数也不知道该传什么。后来仿照写API文档的标准,把每个字段的枚举范围、边界情况、错误码都补全,工具选择的准确率肉眼可见地上来了。不要省这个功夫,工具描述写不好,Agent腿脚就不利索。

5. 常见问题与排查技巧实录

Agent Infra的排查比传统系统难在“不确定性”上。同一个问题,可能模型抽风、上下文污染、工具返回异常、记忆检索偏差,四者都会导致类似的外部表现。我们整理了内部高频踩坑清单,放在这里给大家当速查表。

5.1 上下文污染的典型表现与规避

典型表现是Agent越到后面越“丢三落四”,比如在前面的工具调用中已经拿到订单金额,后面生成总结时却硬说没获取到。排查后发现是上下文太长,早期信息被截断,或者重要性权重太低,模型直接忽略。

规避手段有几种,我按有效性排序:

  • 动态压缩:每轮工具返回结果到达时,立刻用一次轻量模型调用压缩成关键字段摘要
  • 分段控制:上下文按时间线分块,模型可以根据需要回溯指定段落,而不是全部平铺
  • 关键信息置顶:把用户核心诉求、任务目标、关键约束始终放在上下文最靠前的位置

5.2 模型工具选择错乱

模型有时会选错工具,比如查询天气的Agent突然去调用了订机票的工具。排查发现是工具描述里存在模糊相似语义,让模型产生了歧义。

解决思路是工具描述之间做语义隔离。每个工具描述要写清楚边界:这个工具管什么、不管什么、和另一个工具的区别在哪。同时善用枚举值约束,把参数范围卡死,减少模型自由发挥的空间。

5.3 Trace日志缺少关键决策原因

早期我们的Trace只记录“模型调用了哪个工具”,没有记录“为什么调用这个工具”。结果出问题时完全无法还原模型的决策依据,只能瞎猜。

后续我们在Trace里增加了“思路过程”字段,每个关键决策点都把模型当时的理由和候选列表记录下来。代价是多消耗一些Token,但排查效率提升非常明显。对于高复杂度任务,这部分投入完全值得。

5.4 记忆检索结果漂移

向量检索有个现象:输入问题稍微改几个词,检索出来的记忆片段就可能完全不一样,导致Agent有时候想起这个、有时候想起那个,行为不稳定。

我们的应对方法是检索后加一个重排序层,把候选记忆片段先全部取回来,再用模型统一打分排序,只留下最相关的几条。这让记忆引入的波动性大幅下降。

5.5 成本失控

Agent一个任务可能调用十几次模型接口,token消耗比普通对话高一个量级,如果没做控制,月底账单会很吓人。

我们做了三个维度的成本管控:单任务Token上限、缓存优先机制和模型分级。简单任务走小参数模型,复杂任务才启用大参数模型。另外,工具调用结果如果可以直接复用,就不重复请求模型生成。

6. 我的几点实操体会与后续可以扩展的方向

踩了这么多坑之后,我个人的核心体会是:Agent Infra的本质,是把“模型不确定性”和“工程确定性”之间的冲突调和好。模型天生不确定,基础设施的职责就是在这种不确定上建一层可控的壳,让上层业务不至于被模型的随机性拖垮。

另一个深刻的感受是:Agent Infra当前没有标准答案,每个团队都在用自己的方式摸索。巨头们的方案不一定适合中小团队,开源社区的方案也不一定贴合你的业务场景。最好的做法是理解原理、跑通最小实现,然后结合自己业务的真实痛点去叠加能力。

对我们这个跨平台系统项目来说,Agent Infra从零到一跑通之后,后续有几个方向值得深入:

  • 多Agent协作调度:把单Agent扩展成多Agent协同工作的形态,需要一套更完善的任务编排和消息传递机制
  • Agent评估数据回注:把线上评估结果沉淀成数据集,反哺模型微调,形成数据闭环
  • 人机协同审批流:高风险操作引入人工确认节点,这需要在运行时和业务系统之间做更深的集成

最后分享一个小技巧:刚接触Agent Infra的工程师,别急着追新框架、新名词,先把“一条Agent请求从进来到出去,完整经过哪些环节、每个环节可能出什么问题”画清楚。这个基础打牢了,后面不管是换框架还是自研,都不会慌。

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

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

立即咨询