AI Agent项目重构实战：从单体脚本到分层可扩展架构

做AI Agent项目重构做到第二期这次的核心是把AI生成这条链路彻底打散重建同时把整个项目架构从原来的单体脚本式写法重构成一套能支撑多场景、多模型、可扩展的Agent骨架。上一篇我们厘清了老项目的能力边界和改造清单这一篇就直接讲动手过程AI生成怎么重新设计架构怎么一步步拆哪些地方值得砸钱砸时间哪些地方其实是在给自己挖坑。如果你正打算重构一个AI Agent项目或者想把现有的AI生成功能做得更可控、更稳定这篇文章应该能帮你少走不少弯路。我会把重构前后的对比、模块拆分的思路、关键代码的落地方式、以及踩过的坑都摊开来讲尽量做到拿过来就能用。1. 重构动机老项目为什么撑不住了1.1 旧架构的三大痛点重构之前这个Agent项目已经上线跑了大半年功能看起来不少能聊天、能写文案、能生成结构化数据、能调用几个外部接口。但问题也攒了一堆最要命的有三个。第一所有逻辑都在一个服务里AI生成代码和业务代码完全耦合在一起。提示词直接写在业务文件里改一个prompt要重新发布整个服务出问题也不好定位。改一行提示词结果影响了另一个模块的生成效果这种事情发生了不止一次。第二AI生成这条链路没有任何质量保障。调用模型拿到结果就直接用没有格式校验没有失败重试更别说内容审核和后处理了。模型一旦输出畸形JSON或者夹带多余内容向下游传递就是一连串报错。生成的内容偶尔有错别字或逻辑断裂用户感知很差但我们只能人工盯着修。第三模型能力被写死。所有的生成需求都走同一个入口一个模型打天下不支持按场景切换也不支持流式输出。有的场景需要快速响应有的场景需要深度推理老架构完全没法区分对待。想加一个新场景就要复制粘贴一大段代码然后改参数维护成本越来越高。这三件事叠加在一起团队每天早上都要先看一眼生产日志再决定今天干什么。重构的念头其实早就有了但真正推动我下决心的是有一次用户反馈了一张完全跑偏的生成结果而我花了一个下午才在层层包裹的业务代码里找到罪魁祸首。1.2 重构目标与选型思考这次重构定下的目标很明确把AI生成能力从业务代码里剥离出来做成一个独立的、可编排的、能质量自检的生成层再把整个项目从单体脚本式结构重构成接口层—编排层—能力层—记忆层—模型层的分层架构。选型上我没有追新全部采用了当前生态里最稳的组合。语言还是Python主框架用FastAPI异步支持和类型提示都很舒服接入SSE流式输出也很方便。Agent编排没有上重型框架而是自己写了一个轻量调度核心因为我们的场景里自定义逻辑太多现成框架反而要迁就它。模型层做了一层Provider抽象目前同时接入了GPT系列和几个开源模型切换只改配置。记忆层用了向量数据库存长期记忆短期记忆直接用缓存服务两层分开管理互不干扰。这里要说一个很重要的判断重构不是把代码重写一遍而是把对的边界划出来。边界划对了后面加功能是填空边界划错了重构完就是换了个姿势继续乱。2. AI生成能力重建从能生成到生成得好2.1 生成链路的重新设计老项目的生成逻辑是一条直线用户给需求拼好prompt调一次模型接口拿结果返回。这种直线式链路看着简单实际上完全不可控。这次重构我把它改成了六段式链路需求解析 → 生成规划 → 内容生成 → 结构校验 → 质量评估 → 结果修正需求解析负责把用户的模糊输入转成结构化的生成意图比如用户说帮我写一段电商文案风格活泼一点解析层会把风格、长度、平台、目标人群这些参数抽出来。生成规划根据意图决定要不要拆解任务是直接生成还是需要先检索资料避免所有请求都无脑跑一次大模型。内容生成是真正的模型调用环节但在模型调用之前系统已经准备好了完整的生成上下文和输出约束。结构校验针对的是模型输出的格式问题。我们要求大部分生成结果以JSON结构返回模型偶尔会输出残缺JSON或者夹带解释性文字这一层专门负责兜底。质量评估就更进阶一些规则层面的问题交给代码检查语义层面的问题用一个小模型做打分评估两方面结合。如果评估不通过就带着修正意见进入结果修正环节让生成模型自己改自己。链路从直线变成闭环效果提升是肉眼可见的。最直接的变化是生成结果的可用率从原来的七成左右提升到九成五以上很多以前需要人工盯着的低级错误在链路内部就被消化掉了。2.2 提示词工程把模板化变成可编排很多人对提示词工程的理解还停留在把prompt写得花里胡哨。这次重构我最大的体会是提示词工程真正的核心是结构化管理。老项目把整个prompt拼成一大段字符串塞进代码里改的时候要非常小心。重构后我把提示词拆成三层结构底层是系统提示词定义模型的角色和行为准则中间层是场景模板定义某类任务的标准生成流程顶层是变量注入层每次调用时动态填充用户需求、上下文、示例等。为了管理这些模板我搭了一个很轻量的提示词注册中心用配置文件或者数据库存储模板内容。业务代码不再直接拼prompt而是通过一个模板服务按场景ID拉取。这样产品想调话术、改风格根本不用动代码发个配置就生效了。但这里有个容易翻车的点提示词模板化之后不同版本之间的效果差异很难直观对比。所以我在模板层加了一个实验标签机制每次生成请求可以指定模板版本号线上做A/B对比就非常方便。这个设计刚开始觉得多余后来成了我们调整提示词的核心工具谁改了什么模板效果如何全部有数据可查。注意提示词模板不是写得越细越好。有些模板堆了几千字模型反而抓不住重点。我们实践下来系统提示词控制在500字以内场景模板控制在200~500字输出要求的描述用结构化列表说明效果最稳定。2.3 流式输出与回退机制老项目只能等模型生成完了再一次性返回用户体验就是转圈圈。这次重构我把生成接口全面升级为SSE流式输出支持逐字推送。用户看到文字一段段冒出来体感好了不是一点半点。但流式输出带来一个新的问题如果中途模型报错或者网络抖动用户已经看到一半内容了怎么处理我采用的是流式优先、失败回退的策略。正常情况走SSE逐字推送给前端一旦流中断后端会尝试重连一次重连也失败就切换为同步调用的兜底模式把已生成的内容拼接好再完整返回一次。对用户来说最多是感知到一次轻微卡顿不会看到一半就断掉。流式输出的工程细节也不少。首先是超时管理模型接口和网络请求的读超时、连接超时、空闲超时要分开设置不能一个参数控全局。其次是缓冲机制不是每个token都立刻推给前端而是攒一小批统一推送减少网络开销。最后是心跳包长时间没有新token时要主动发一个保持连接的信号避免前端误判连接已断开。这些细节看起来不起眼但在生产环境里每一个都能坑你一次。3. 架构重建Agent的核心骨架3.1 模块拆解接口层、编排层、能力层、记忆层、模型层架构重建的基本思路是做分层和解耦。现在项目从底往上分五层模型层、记忆层、能力层、编排层、接口层。模型层是所有大模型调用的统一入口。不管底层是GPT、开源模型还是未来的新模型对外暴露的都是同一个接口传入消息列表和参数返回生成结果。每次调用记录token消耗、耗时、模型版本这些数据后面用来做成本分析和质量追踪。模型层还做了降级策略主模型不可用时自动切换到备用模型。记忆层分两块短期记忆存在缓存服务里保存当前会话的上下文长期记忆存在向量数据库里保存用户的偏好、历史关键信息和跨会话的知识。短期记忆和长期记忆的读写都通过统一接口上层不需要关心数据到底存在哪里。能力层是Agent执行具体操作的集合包括内置能力如内容生成、文本总结、信息抽取和外部工具如查数据库、调外部API、读取文件。每个能力都有独立的描述文件和参数Schema形成一个能力注册中心Agent可以按需发现和调用。编排层是Agent的大脑负责理解用户意图、拆解任务、选择能力、调度执行。它本身不干具体活而是像项目负责人一样分配工作。最后是接口层对外提供HTTP接口、WebSocket接口和Webhook接口把Agent能力封装成服务。分完层之后最直观的变化是改动一个层不影响其他层调试和排查问题的效率提升非常明显。3.2 上下文管理从全都塞进去到按需取用上下文管理是最容易被低估的模块但它的地位实际上决定了Agent能力的上限。老项目把所有对话历史一股脑塞给模型上下文一长模型要么丢失早期信息要么开始胡言乱语而且成本直线上升。这次重构我把上下文管理做成了三层结构原始窗口、压缩摘要、长期记忆。原始窗口保留最近几轮对话的完整内容保证当前对话的连贯性。当对话超过窗口长度就把更早的历史交给摘要模型做一轮压缩生成一段简洁的要点摘要放回到上下文里。这样无论对话多长模型看到的始终是摘要近期原文不会上下文爆炸。长期记忆则按需注入不把用户所有的历史记录全塞进来而是根据当前意图做相似度检索只提取相关片段。比如这个用户三个月前说过自己喜欢极简风格今天让他设计一个页面系统会从长期记忆里把偏好极简风格这条检索出来注入上下文模型就能给出更贴合口味的方案。我在实现这套机制时最深的感受是短期窗口和长期记忆的配合必须设计好冷热分层热数据放短期窗口温数据放压缩摘要冷数据放长期记忆哪个数据该放哪层要有清晰的判断逻辑。3.3 Agent编排器与技能机制实现思路编排器说出来不神秘本质上就是一个带循环的决策器。它的工作流是接收用户请求调用规划模型决定今天这个任务要不要拆解、拆成几步、每步用什么能力然后按照计划逐步执行每执行一步都检查结果该终止就终止。实现时我做了两层循环控制。外层是大任务的主循环负责拆解和步骤推进内层是单步执行的子循环负责重试、自我修正和结果校验。这样设计的好处是每一步都可以独立控制不会因为某一步小报错导致整个任务失败。还有一个很关键的细节是最大步数限制。Agent在自由调用工具时很容易陷入死循环同一个错误修正来修正去出不来我设置了默认最大步数我们生产环境设的是10步超过就强制结束并返回已完成的中间结果。技能机制是同步引入的。我借鉴了当前Agent社区通用的SkillMemoryMCP的思路把Agent能做的事情拆成一个个独立注册的技能。每个技能包含功能描述、参数Schema、执行函数三个组成部分。Agent需要调用某个能力时先根据技能描述判断哪个技能匹配生成参数后调用执行函数。外部工具通过MCP协议接入这样新工具接入不需要改Agent本身只需要新增一个MCP服务并注册技能说明。这套机制跑起来之后加新功能成了一件很填表的事情开发效率提升非常明显。4. 核心代码落地关键实现片段4.1 模型层统一接入的实现模型层是所有上层模块的地基设计上要做到换模型不改业务代码。我用抽象基类定义统一的生成接口再为每个模型写一个适配实现。from abc import ABC, abstractmethod from typing import AsyncIterator, Dict, List, Optional, Any class LLMProvider(ABC): 所有模型Provider的统一抽象接口 abstractmethod async def generate( self, messages: List[Dict[str, str]], tools: Optional[List[Dict[str, Any]]] None, temperature: float 0.7, max_tokens: int 2048, response_format: Optional[str] None, ) - Dict[str, Any]: 同步生成返回完整结果 pass abstractmethod async def stream_generate( self, messages: List[Dict[str, str]], tools: Optional[List[Dict[str, Any]]] None, temperature: float 0.7, max_tokens: int 2048, ) - AsyncIterator[str]: 流式生成逐段返回文本增量 pass abstractmethod def model_name(self) - str: 返回当前Provider使用的模型名称 pass有了这个抽象接口上层调用完全屏蔽了模型差异。我用类似工厂模式的方式管理Provider实例通过配置项决定当前默认使用哪个模型。PROVIDER_REGISTRY: Dict[str, type] {} def register_provider(name: str): def decorator(cls): PROVIDER_REGISTRY[name] cls return cls return decorator def get_provider(name: Optional[str] None) - LLMProvider: provider_name name or settings.default_provider provider_cls PROVIDER_REGISTRY[provider_name] return provider_cls(api_keysettings.get_api_key(provider_name))这种写法看起来简单但对后续扩展的意义非常大。后面接新的模型或者新的API服务只需要写一个新的Provider类注册进去其他模块一行代码都不用动。实操心得Provider层一定要把token用量、耗时、错误信息都记录到日志。上线后你会发现模型层的可观测性直接影响你排查问题的速度。4.2 编排器核心循环的实现编排器的核心是一个有限循环每一轮做三件事调用规划模型决定下一步动作根据动作执行对应技能把执行结果反馈给模型继续决策。我写了一个简化的实现来说明思路。class AgentExecutor: def __init__(self, max_steps: int 10): self.max_steps max_steps self.skill_registry SkillRegistry() async def run(self, task: str, context: AgentContext) - AgentResult: messages self._build_initial_messages(task, context) steps [] for step in range(self.max_steps): # 1. 调用规划模型决定下一步动作 response await self.provider.generate(messages) action self._parse_action(response) # 如果模型判定任务完成跳出循环 if action.type finish: return AgentResult(successTrue, stepssteps, outputaction.output) # 2. 根据动作执行技能 if action.type call_skill: skill self.skill_registry.get(action.skill_name) if not skill: messages.append(agent_message( f技能 {action.skill_name} 不存在请换一个可用的技能 )) continue result await skill.execute(**action.arguments) steps.append(action) # 3. 把执行结果反馈给模型 messages.append(tool_message(result)) elif action.type error: return AgentResult(successFalse, stepssteps, erroraction.error) # 超过最大步数时强制结束 return AgentResult(successFalse, stepssteps, errormax_steps_exceeded)这段代码是我删掉了大量异常处理和日志记录后的简化版但整体结构就是这样的。好用的编排器不在于逻辑多复杂而在于每一步都清晰可追踪。我们把每一步的动作、参数、结果、耗时全部记录到结构化日志里回放一个任务的执行过程非常直观这也是排查Agent诡异行为的基础。4.3 技能注册与执行机制技能注册我用了一个装饰器实现代码非常简洁但使用体验很好。class SkillRegistry: def __init__(self): self._skills: Dict[str, Skill] {} def register(self, name: str, description: str, parameters_schema: dict): def decorator(func): self._skills[name] Skill( namename, descriptiondescription, parameters_schemaparameters_schema, handlerfunc, ) return func return decorator def get(self, name: str) - Optional[Skill]: return self._skills.get(name) def list_skills(self) - List[dict]: return [ {name: s.name, description: s.description, parameters_schema: s.parameters_schema} for s in self._skills.values() ] # 使用示例 registry SkillRegistry() registry.register( namefetch_user_profile, description根据用户ID获取用户的画像信息包含偏好、历史订单等, parameters_schema{ type: object, properties: {user_id: {type: string}}, required: [user_id], } ) async def fetch_user_profile(user_id: str): # 实际执行逻辑 return await user_service.get_profile(user_id)技能描述写得好不好直接影响模型能不能正确选择技能。我实践下来的经验是技能描述要包含什么时候该用、什么时候不该用这两段信息比单纯写获取用户画像这种干巴巴的描述效果好很多。比如加上当用户询问订单状态或历史记录时使用当用户只是闲聊时不要使用这样的边界说明模型的选择准确率会明显提升。5. 实操中的坑与排查技巧5.1 高频问题速查表重构过程中遇到的问题是海量的我从中筛了几个出现频率最高、也最容易被忽视的整理成了一张速查表方便遇到同类问题时直接翻阅。症状根本原因解决方案生成结果偶尔抛出JSON解析异常模型输出夹带说明文字或JSON截断在提示词中强制只输出JSON增加结构校验层自动修复上下文一长效果急剧下降把全部历史对话塞给模型窗口溢出引入摘要压缩机制只保留近期原文加早期摘要Agent反复调用同一个技能出不来缺少迭代中止条件模型陷入自我纠错循环设置最大步数限制超过阈值强制返回中间结果流式输出偶尔断流用户看到半截内容网络抖动导致连接断开缺少回退策略实现自动重连降级为同步返回的兜底方案切换模型后输出风格大变不同模型对同一套提示词的响应差异大为每个模型单独维护一套提示词模板版本隔离多个技能并行调用时偶发超时没有做并发控制下游接口被压垮引入信号量限制最大并发数设置单技能调用超时改造后首字延迟反而变高请求链路变长各环节排队耗时叠加对高频路径做缓存预热独立测量每层耗时优化瓶颈模版上线后效果不如预期模板变更没有做对比评估凭感觉上线利用模板版本标签做A/B对比数据说话再决定是否全量这张表我自己贴在工位上遇到问题直接查不用每次从头排查。5.2 重构过程中最值得分享的心得重构结束之后回头复盘有几点体会特别想分享给做类似项目的人。第一架构重建一定要和功能开发分开进行。我这轮重构中途差点又陷入顺便加个新功能的诱惑里还好及时刹住了。重构期间只做等价替换老功能能跑通就算成功一切新增功能等重构结束再排期。混在一起的结果往往是重构没做好新功能也不稳定两边都焦头烂额。第二可观测性要前置。以前的日志只记录成功和失败具体过程像黑盒。这次重构我一开始就给每一步加上trace_id从用户请求进来到编排器决策、技能调用、模型生成整条链路全程标记。出问题的时候按trace_id一拉哪一步慢了、哪一步失败了清清楚楚。强烈建议在写架构代码之前先把日志规范定下来。第三提示词模板的版本管理是刚需。一开始我用配置文件管理模板后来发现总有人直接改线上配置改完还不记录。后来我要求所有模板变更必须通过代码提交配合自动生成的版本记录每一行提示词改动都有迹可循。这个改动一度被认为多此一举直到有一次线上生成质量异常靠版本对比十分钟就定位到是哪次改动引入的问题大家才认可了这个机制。第四模型选择不要追求一步到位。整个重构过程我前后试了七八种模型组合每种模型在不同场景下的表现差异真的很大。有的模型擅长推理有的模型响应快有的模型输出格式稳定。最终的方案不是选一个万能模型而是按场景路由重推理任务走强模型轻量任务走快模型成本和质量都兼顾了。6. 后续还可以怎么扩展重构的阶段性目标是达成了但Agent项目的事情永远做不完。我已经在规划下一步的几个方向顺手也分享给大家做个参考。短期记忆这一层我准备接入一个轻量级的对话摘要缓存把对话中反复出现的实体信息比如用户的名字、偏好、项目代号抽出来存成结构化标签下次对话直接查标签不用再做全文相似度检索。现在记忆层按需检索的效果还不错但碰到表达方式差异很大的查询时召回率还有提升空间。编排器这边我想把当前的线性循环改成支持子任务级联的树形结构。现在遇到一个非常复杂的任务拆成多个步骤之后是串行执行的某些步骤之间其实没有严格依赖完全可以并行。树形编排加上并行执行理论上可以把复杂任务的耗时压缩一半。这个改造涉及的地方不少要动编排器的核心数据结构得排一个单独的迭代来做。对外接口方面我也在考虑把Agent能力开放成一套可配置的Webhook服务。现在很多用户想在自己的系统里直接调用Agent能力但没有技术背景又搭不起来完整环境。如果提供一套只需填参数就能创建自定义Agent的配置界面门槛能降下来不少。这算是从自己用到给别人用的跨越工程量和交互设计都要重新考虑。还有模型端的进展值得盯一下。开源模型的迭代速度太快了这个季度测试还不太行的模型下个季度可能就变成了性价比最优解。我的模型层做了Provider抽象换模型本来就是改配置的事但更要关注的是新模型带来的能力边界变化——以前需要靠复杂的任务拆解才能完成的事情新模型可能一步就能生成这时候编排策略也要跟着调整。最后再分享一个这轮重构教会我的事Agent项目的架构重构真正考验的不是写代码的能力而是判断哪些东西值得抽象、哪些东西宁愿写死的取舍能力。抽象过度系统变得臃肿难懂写死太多后续改造成本飙升。这个平衡点需要在一次次迭代里慢慢摸我给自己的原则是当你第三次遇到同样的重复代码时才是抽公共模块的正确时机前两次先老老实实允许自己写得丑一点。