两个月前，我搭了一个 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, tools=None): 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"}, json=payload, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]这段代码里没有任何业务逻辑，只负责围绕 OpenAI 兼容协议做“输入消息、输出回复”。好处是后面所有环节都能在这个基础上加：加记忆、加工具、加多轮循环。
2.2 让 Demo 有“记忆”：上下文管理的最小实现
很多人做聊天机器人，第一步就是把所有历史消息全塞给模型。这个做法在对话轮次少的时候没问题，但一旦聊得久了，token 会暴涨，模型注意力也会被无关内容干扰。
我最初的实现是一个 Session 类：
class Session: def __init__(self, max_tokens=4000): 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 Calling
Demo 真正发生质变的时刻，是我给模型接上了“工具调用”。也就是 Function Calling。简单说，就是模型不再直接回答一个它不知道的问题，而是先请求调用某个函数，拿到函数执行结果后，再基于真实数据生成回复。
还是拿天气举例。我定义了这样一个工具：
def get_weather(city: str): # 真实场景这里接入天气 API，demo 阶段先返回固定值 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_turns=3): for _ in range(max_turns): message = chat(history, tools=TOOLS) 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_ascii=False), }) 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, tools=None, **kwargs): pass class OllamaClient(BaseModelClient): def __init__(self, base_url, model): self.base_url = base_url self.model = model def chat(self, messages, tools=None, **kwargs): # 复用原来的 requests 逻辑 pass class CloudAPIClient(BaseModelClient): def __init__(self, api_key, model): self.api_key = api_key self.model = model def chat(self, messages, tools=None, **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_ascii=False) 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 只是写配置，而不是写代码。
