ARTICLE DETAIL

资讯详情

深耕商务建站与企业官网运营的一线实战洞察。

Agent Infra实战:从零搭建企业级Agent基础设施的关键技术

Agent Infra实战:从零搭建企业级Agent基础设施的关键技术 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_seconds10): try: return await asyncio.wait_for( tool_func(), timeouttimeout_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(fagent:{trace_id}:ctx, f{role}:{content}) async def load_relevant_long_term(self, user_id, query, top_k5): query_vector self.vector.embed(query) results await self.vector.search( indexagent_memory, query_vectorquery_vector, top_ktop_k, filter{user_id: user_id} ) return results async def finalize_memory(self, trace_id, user_id): # 拉取整条短期记忆 raw_ctx await self.redis.lrange(fagent:{trace_id}:ctx, 0, -1) # 调用摘要模型提炼长期记忆 summary await self.summarizer.summarize(raw_ctx) await self.vector.upsert( indexagent_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(fagent:{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请求从进来到出去完整经过哪些环节、每个环节可能出什么问题”画清楚。这个基础打牢了后面不管是换框架还是自研都不会慌。
返回列表
PREV
查看更多资讯
NEXT
返回资讯列表