ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

从AI Demo到Agent平台:架构分层与工程化实践

从AI Demo到Agent平台:架构分层与工程化实践 两个月前我搭了一个 AI 对话 Demo核心功能就是和大模型聊聊天顺便能按模板回答几个行业问题。当时觉得挺成功周围朋友都说有意思。但等我把它拿到真实业务场景里被连续问到“能不能帮我写一份周报”“能不能查一下上周的销售数据”“能不能接入我们的工单系统”我才意识到一个能跑通的 AI Demo 和一套真正可演进的 Agent 平台中间隔着的不是几行代码而是整套工程化思维。这篇文章是这个系列的第一篇我会先讲清楚“为什么从 Demo 开始”“Demo 和 Agent 平台到底差在哪”再拆解我是如何一步步把单体脚本改造成分层架构并给出实测中的踩坑记录。适合两类人看一类是刚入门 AI 应用开发、手里有个聊天 Demo 想往上走的同学另一类是已经在做 Agent 平台想看看别人怎么理解“可演进”这三个字的工程师。1. 先说清楚为什么从 Demo 写起1.1 这个系列的起点我那个只能聊天的 Demo我最初的 Demo 很简单Python 脚本调本地模型Flask 包一个接口前端就是一个输入框加一个发送按钮。模型用的本地部署的 Qwen 系列通过 Ollama 暴露成 OpenAI 兼容接口代码总共不到两百行。会话记忆存在一个全局列表里用户每次提问就把整个聊天历史塞给模型token 超了就粗暴地从最前面截断。这个版本跑起来很容易效果也还行。演示给朋友看大家夸一句“有点意思”。但问题恰恰出在“有点意思”这四个字上。因为 Demo 的本质是验证一个点模型能不能在这个场景下给出合理回复。而平台要解决的是另一个问题在不同用户、不同任务、不同工具之间如何稳定、可维护、可观测地完成任务。如果一直停留在 Demo 阶段代码写得再顺手也只是在“证明能跑”而不是在“交付价值”。1.2 Demo 和 Agent 平台的差距不是量的区别是质的区别我用一张表整理过两者的差别后来每次团队讨论架构我都会先让大家重新看一遍这张表维度AI 对话 Demo可演进 Agent 平台场景目标验证模型效果解决业务问题上下文内存里的列表持久化、分级记忆工具调用没有或硬编码可注册、可编排、可审计模型固定一个模型多模型可切换、可 A/B可靠性演示为主有评测、有回归、有兜底可观测性print 日志trace、成本、效果指标演进方式推倒重写替换模块、增量扩展这张表不是说 Demo 没用。恰恰相反如果没有 Demo 阶段的试错我不会知道用户真正在意什么比如他们希望 Agent 能主动调用接口查数据而不是只凭大模型的“记忆”瞎编他们希望对话能连续而不是问两句就忘了前文。所以从 Demo 到平台不是把 Demo 代码变大而是把“能跑”变成“能持续跑、能扩展、能上线”。2. 从 Demo 出发先做出一版能跑通的最小系统2.1 技术选型为什么我用 OpenAI 兼容接口我坚持用 OpenAI 兼容接口不是因为某个厂商有多好而是因为它已经成为事实上的标准接口。本地部署的 Ollama、vLLM云端各家大模型服务基本都兼容或提供转换层。选用这个接口意味着以后换模型、换部署方式业务代码几乎不用动。我的最小 Demo 大概是这个形态import requests OLLAMA_BASE http://localhost:11434/v1 MODEL qwen2.5:7b def chat(messages, toolsNone): payload { model: MODEL, messages: messages, temperature: 0.7, } if tools: payload[tools] tools resp requests.post( f{OLLAMA_BASE}/chat/completions, headers{Content-Type: application/json}, jsonpayload, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message]这段代码里没有任何业务逻辑只负责围绕 OpenAI 兼容协议做“输入消息、输出回复”。好处是后面所有环节都能在这个基础上加加记忆、加工具、加多轮循环。2.2 让 Demo 有“记忆”上下文管理的最小实现很多人做聊天机器人第一步就是把所有历史消息全塞给模型。这个做法在对话轮次少的时候没问题但一旦聊得久了token 会暴涨模型注意力也会被无关内容干扰。我最初的实现是一个 Session 类class Session: def __init__(self, max_tokens4000): self.history [] self.max_tokens max_tokens def add(self, role, content): self.history.append({role: role, content: content}) self._trim() def _trim(self): total sum(len(msg.get(content) or ) for msg in self.history) while total self.max_tokens and len(self.history) 2: self.history.pop(0) total sum(len(msg.get(content) or ) for msg in self.history)这个方案很粗糙核心思想就一句话保留最近的对话丢掉太旧的内容。它能解决“token 爆炸”的燃眉之急但不解决“我们要记住用户偏好”的问题。后面我会专门讲分层记忆这里先让 Demo 能连续对话就行。2.3 跨过第一道坎加入 Function CallingDemo 真正发生质变的时刻是我给模型接上了“工具调用”。也就是 Function Calling。简单说就是模型不再直接回答一个它不知道的问题而是先请求调用某个函数拿到函数执行结果后再基于真实数据生成回复。还是拿天气举例。我定义了这样一个工具def get_weather(city: str): # 真实场景这里接入天气 APIdemo 阶段先返回固定值 return {city: city, weather: 晴转多云, temperature: 24} TOOLS [ { type: function, function: { name: get_weather, description: 查询某个城市的当前天气, parameters: { type: object, properties: { city: {type: string, description: 城市名比如杭州} }, required: [city], }, }, } ]调用循环如下def run_agent(history, max_turns3): for _ in range(max_turns): message chat(history, toolsTOOLS) history.append({ role: assistant, content: message.get(content), tool_calls: message.get(tool_calls), }) if not message.get(tool_calls): return message[content] for tc in message[tool_calls]: func {get_weather: get_weather}[tc[function][name]] result func(**json.loads(tc[function][arguments])) history.append({ role: tool, tool_call_id: tc[id], content: json.dumps(result, ensure_asciiFalse), }) return 抱歉我还没能在规定步骤内解决这个问题。这段代码直接决定了 Agent 的雏形不是“问一句答一句”而是“模型自主决策要调哪个工具 → 系统执行 → 把结果喂回去 → 模型继续推理”。我实测下来加入这一层之后用户满意度比之前翻了一倍都不止。因为模型不再只会“一本正经地胡说八道”而是真的连接到外部数据了。3. 为什么 Demo 撑不起 Agent 平台架构分层的思考3.1 核心转变从“一条流程”到“一组模块”Demo 的代码是线性流程接收消息 → 塞入历史 → 调模型 → 返回结果。这个流程在单一场景下没问题但一旦要接入多个 Agent、多个工具、多个模型线性流程就变得不可维护。我重构时把系统按职责拆成了几层模型接入层封装不同模型来源统一请求入口。工具与插件层负责工具的注册、执行、鉴权。记忆层管理短期会话、长期知识、摘要记忆。编排层也叫 Agent Harness负责循环、步数控制、决策路由。应用与 API 层面向用户和外部系统。可观测层记录 trace、成本、效果指标。这个拆分不是我拍脑袋想出来的而是当我开始同时维护“查天气 Agent”“周报 Agent”“数据问答 Agent”三个场景后自然被逼出来的。因为没有统一分层每加一个 Agent我就要复制一份对话循环代码改起来极其痛苦。3.2 可演进的关键Agent 本身也要“配置化”代码分层的下一步就是把 Agent 的定义从 Python 代码中抽出来。一个 Agent 用什么模型、用什么系统提示词、能调哪些工具、最多跑几步、超时怎么处理这些都应该用配置文件描述而不是写死在代码里。我现在的 Agent 定义长这样id: weekly_report_agent model: qwen2.5:7b system_prompt: | 你是周报助手。根据用户提供的素材整理成结构清晰的周报。 tools: - search_notes - get_calendar memory: type: session window: 20 max_steps: 5 on_timeout: ask_user_to_follow_up这样做的好处很直接新加一个 Agent 不需要改平台代码只需要加一个 YAML 文件。产品经理也能通过后台界面配置系统提示词和工具列表而不需要麻烦开发工程师。这也是我理解的“可演进”的核心不是把代码写得多高级而是把变化的部分变成数据把稳定的部分变成平台。3.3 Skill 与 Agent 的边界我踩过的概念误区规划架构时我一直分不清 Skill 和 Agent 到底是什么关系。后来我用一句话理清了Skill 是“能力原子”Agent 是“决策主体”。Skill 是某个可复用的能力比如“生成周报模板”“查询数据库”“调用天气 API”。它没有目标只有操作方式。Agent 则带着一个目标它能根据用户意图决定何时调用哪个 Skill还能在调用失败后调整策略。举个例子周报 Agent 需要调用search_notes搜索笔记和get_calendar读取日历这两个 Skill。Skill 本身不知道用户想干嘛但 Agent 知道用户说“帮我写周报”Agent 就会先查日历再搜笔记最后结合这两部分素材生成周报。如果把 Skill 写死在 Agent 逻辑里那每加一个能力都要动 Agent 代码。把 Skill 和 Agent 解耦之后我只需要新增 Skill然后在某个 Agent 的配置里把 Skill 名加进tools列表即可。4. 迈向平台从最小系统到可扩展架构的具体实施路径4.1 第一步把模型接入层抽象成可插拔接口我把原来的chat()函数改成了抽象类from abc import ABC, abstractmethod class BaseModelClient(ABC): abstractmethod def chat(self, messages, toolsNone, **kwargs): pass class OllamaClient(BaseModelClient): def __init__(self, base_url, model): self.base_url base_url self.model model def chat(self, messages, toolsNone, **kwargs): # 复用原来的 requests 逻辑 pass class CloudAPIClient(BaseModelClient): def __init__(self, api_key, model): self.api_key api_key self.model model def chat(self, messages, toolsNone, **kwargs): # 调用云端模型接口 pass这样改完之后业务层代码只依赖BaseModelClient这个抽象。我可以在不同模型之间切换也可以做 A/B 测试同一批问题让两个模型各跑一遍然后对比效果。这个抽象带来的价值在实际运行一周后体现得非常明显——某个模型版本升级后效果回退我直接切回旧模型业务完全不受影响。4.2 第二步搭建工具注册中心工具管理不能靠“在代码里维护一个大字典”。我采用装饰器模式做注册中心TOOL_SCHEMAS [] TOOL_FUNCTIONS {} def tool(name, description, parameters): def decorator(func): TOOL_SCHEMAS.append({ type: function, function: { name: name, description: description, parameters: parameters, }, }) TOOL_FUNCTIONS[name] func return func return decorator tool(get_weather, 查询指定城市的天气, { type: object, properties: {city: {type: string}}, required: [city], }) def get_weather(city: str): return {city: city, weather: 晴}执行工具时直接从TOOL_FUNCTIONS里取函数def execute_tool(name, args_json): func TOOL_FUNCTIONS[name] try: result func(**json.loads(args_json)) return {ok: True, result: result} except Exception as e: return {ok: False, error: str(e)}这里必须强调一点工具是有权限的。我把工具分成了两级只读类查天气、查数据库 SELECT可以直接执行写操作类发邮件、改数据库、删除文件必须加一层人工确认。否则 Agent 一旦被恶意提示词“诱导”可能造成不可挽回的损失。4.3 第三步把记忆从“变量”升级为“服务”Demo 阶段记忆就是一个Session类进程重启就丢。平台阶段记忆至少分三层短期会话记忆同一会话内的上下文需要持久化到 Redis 或数据库。长期记忆用户偏好、历史结论可以写入向量数据库按需检索。摘要记忆当会话太长时把旧内容总结成几条要点继续保留摘要丢弃细节。我当时实现了一个简单的摘要逻辑def summarize(history): prompt 请用三句话概括以下对话中已经确认的信息不要输出无关内容\n json.dumps(history, ensure_asciiFalse) reply chat([ {role: system, content: 你是记忆整理助手。}, {role: user, content: prompt}, ]) return reply[content]虽然这个方案每次都会消耗一次模型调用但换来的是长会话场景下的稳定表现。比如用户连续问五个问题前面四个细节都被压缩成三句话第五个问题到来时模型依然能准确引用前文信息而不是被 4000 token 的噪音淹没。4.4 第四步引入可观测性与评测集平台没有日志和指标就是裸奔。我给每个关键动作都打了结构化日志写 JSON Lines 文件后续直接导入数据库{event: tool_call, agent: weekly_report_agent, tool: get_calendar, latency_ms: 231, timestamp: 2025-06-01T10:00:00Z}同时我建了一个很小的评测集用来回归测试 Agent 变更cases: - name: 查天气 input: 杭州明天适合出门吗 expected_steps: [get_weather] expected_contains: [天气] - name: 写周报 input: 根据我这两天的记录写周报 expected_steps: [search_notes, get_calendar] expected_contains: [本周]每次改完系统提示词或模型参数我先跑一遍评测集看哪些用例挂了再决定要不要发布。虽然这套评测还很原始但它已经帮我拦住过一次“升级系统提示词后周报 Agent 不调用日历工具”的回归问题。5. 实操中踩过的坑与排查方法5.1 上下文越长效果越差吐出旧内容或答非所问这是刚开始最容易踩的坑。会话历史长了以后模型容易“迷失在长文本里”。我的排查方法分三步先看是不是 token 数量超出模型上下文窗口的一半再看是不是旧内容里混杂了无关信息最后决定是截断还是摘要。目前最稳妥的方案是“摘要 滚动窗口”结合超过窗口的部分定期压缩成摘要当前窗口只保留最近几轮完整对话。这个方案牺牲了一点信息量换来了稳定性和低成本。如果你对长对话的完整性要求很高可以考虑向量数据库但这是后话。5.2 Function Calling 返回结果不稳定我用本地 7B 模型时Function Calling 偶尔会不按标准格式返回比如缺了tool_calls字段或者 JSON 参数解析失败。我的兜底策略有两个一是给模型明确的系统提示词比如“如果用户问题涉及查询天气必须调用 get_weather”。二是调用执行层加 try-except解析失败时把错误信息回传给模型让模型自己纠正def safe_execute(func, args_json): try: result func(**json.loads(args_json)) return {ok: True, result: result} except Exception as e: return {ok: False, error: str(e)}实测下来把错误信息作为 tool 消息回传后大部分模型都能在下一轮自行纠正调用参数。5.3 多 Agent 一多就开始“踢皮球”单一 Agent 跑得挺好但我一接入多个 Agent它们之间就可能无限循环A 说这个归 B 管B 说你还是找 A 吧。我的解法是用一个 Supervisor Agent 做路由并在编排层限制最大步数。Supervisor 只负责判断“用户意图属于哪个领域”然后分发给具体 Agent。分发后具体 Agent 只处理自己领域内的问题不允许跨域推诿。同时我把所有 Agent 的max_steps都设置为 3 到 5。超过步数就主动向用户说明“需要人工介入”。这样能避免系统在一个无法收敛的问题上空转。5.4 本地小模型和商业模型的取舍很多人以为本地部署免费又安全就一股脑全用本地模型。实际跑下来7B 模型的推理能力、工具调用稳定性和商业大模型差距还是明显的。我的建议是混合路由场景推荐模型原因简单问答、闲聊本地小模型成本低、响应快复杂推理、多步工具调用商业模型准确率高涉及隐私数据本地模型或私有化部署数据不出域对延迟敏感本地模型省去网络开销混合路由听起来复杂但实现起来就是一个配置文件不同 Agent 的model字段填不同的模型 ID 即可。5.5 常见问题速查表问题现象常见原因解决方向答非所问模型回复与问题无关上下文过长、提示词冲突摘要、窗口裁剪、精简 system prompt工具调用乱调错工具或参数异常小模型指令遵循弱加提示、加校验、换更强模型Agent 循环多 Agent 之间互相踢皮球缺少路由、步数无上限Supervisor 分发、限制 max_steps效果回归升级后某个场景变差模型版本、提示词、工具变化维护评测集发布前回归成本失控token 消耗过高历史无限堆积、循环调用摘要压缩、限制步数、缓存结果6. 下一步计划这个系列的第一篇就写到这里。接下来我打算把当前这个轻量 Agent 平台开源出来做成一个 FastAPI 服务内置模型接入层、工具注册中心、配置化 Agent 定义和基础日志模块。后续几篇会依次展开记忆服务的完整设计、Supervisor 多 Agent 路由的具体代码实现、以及基于评测集的自动回归流程。如果你现在也卡在“Demo 能跑但不知道下一步怎么办”的阶段我最大的体会是不要急着堆功能先把模型接入层、工具层、记忆层这三块抽象出来。这三块一旦稳定后面加新 Agent 只是写配置而不是写代码。
返回列表