AI Agent编程学习系列」第 2篇：Agent架构五层模型,从LLM内核到交互界面——用TaoToken统一Key跑通分层调用链

发布时间：2026/10/9 23:22:18

1. 为什么你的 Agent 代码写到 500 行就崩了我见过太多 Agent 项目死在同一个地方一个agent.py文件里塞了模型调用、工具函数、对话历史、while 循环、还有 Flask 路由。刚开始跑得挺欢加第三个工具的时候开始出 bug加第五个工具的时候自己都不敢改了。这不是你代码水平的问题是架构缺了一张地图。Agent 系统本质上是个多层协作的软件工程问题它至少包含五个职责完全不同的域模型推理、工具执行、记忆存取、任务编排、用户交互。把这五件事揉在一起写等于把数据库查询、HTTP 路由、业务逻辑全写进一个函数——能跑但没法维护。五层模型就是给 Agent 画一张分层地图层级名称核心职责典型技术L1模型层LLM 推理与响应生成GPT、Claude、Qwen、DeepSeekL2能力层工具定义、注册与执行Function Calling、MCP、APIL3记忆层短期与长期记忆管理上下文窗口、向量数据库L4编排层任务规划、调度与执行控制ReAct、Plan-and-ExecuteL5交互层用户界面与输入输出处理Web、CLI、Bot分层之后你能得到什么最直接的好处是替换成本骤降。想把 L1 从 GPT 换成 Claude只改模型层的实现类想给同一个 Agent 核心加个 Web 界面只写一个新的 L5L1-L4 一行不动。这就是关注点分离带来的工程红利。但分层架构有个绕不开的前置问题每一层都要调模型每一层都要配 Key。L1 要调推理接口L4 编排时要做意图识别L3 记忆压缩也要调模型做摘要。如果每层各配一套 Key、各写一套 base_url你的配置文件会比代码还乱而且排查问题时根本不知道是哪层的请求挂了。这篇的做法是用 TaoToken 的统一 Key 和统一 API 通道把五层的模型调用收敛到一个入口。这样你只需要维护一份配置五层共享同一个通道链路可观测、边界清晰。下面从环境准备开始逐层落地。2. TaoToken 统一 Key 前置一份配置喂饱五层在动手写五层代码之前先把模型调用的通道统一掉。这一步做对了后面每一层都省事。TaoToken 在这里扮演的角色是统一的模型接入层你拿到一个 API Key配一个 Base URL就能在五层里用同一套凭证调用不同的模型。L1 用强模型做推理L3 用便宜模型做记忆摘要L4 用快模型做意图分类——模型可以不同但 Key 和通道是同一个。先拿 Key。访问控制台创建 API Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite创建完 Key 之后记下两个东西API Key 本身以及 Base URL。Base URL 统一用https://taotoken.net/api注意这个地址后面不加 UTM 参数它是真正的接口端点。控制台和文档页才带归因参数。接下来是模型 ID。五层里我会用到三个模型档位你在配置里按需替换推理档L1 主力claude-sonnet-4-5或gpt-4o快档L4 意图分类gpt-4o-mini或qwen-turbo摘要档L3 记忆压缩gpt-4o-mini模型 ID 的具体可用列表以文档为准接入方式看这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite现在把配置写成一个独立的config.py五层都从这里读不要在各层里硬编码# config.py import os TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, sk-your-key-here) TAOTOKEN_BASE_URL https://taotoken.net/api # 五层共用的模型档位 MODEL_REASONING claude-sonnet-4-5 # L1 推理 MODEL_FAST gpt-4o-mini # L4 意图分类 MODEL_SUMMARY gpt-4o-mini # L3 记忆摘要 # 统一客户端参数 DEFAULT_TIMEOUT 60 MAX_RETRIES 2如果你用的是 OpenAI 兼容的 SDK客户端初始化长这样from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL client OpenAI( api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL, )这里有个关键点五层共用同一个 client 实例。不要每层各 new 一个那样连接池、重试策略、超时配置都会分散出问题时你没法统一调整。把 client 作为依赖注入到各层这也是后面create_agent工厂函数要做的事。如果你更习惯用 Claude Code 或 Cline 这类工具做开发辅助它们的配置也是同一套逻辑——Base URL 填https://taotoken.net/apiKey 填你创建的那个Model ID 填上面选的档位。三件套齐了就能跑。Coding Plan 适合长期编码场景配置入口在这里https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite前置做完你的项目结构应该是这样agent_project/ ├── config.py # 统一配置 ├── l1_model.py # 模型层 ├── l2_tools.py # 能力层 ├── l3_memory.py # 记忆层 ├── l4_orchestrator.py # 编排层 ├── l5_interface.py # 交互层 └── main.py # 组装入口每个文件只干一件事。下面逐层写。3. 五层可复制配置从 L1 到 L5 的完整代码这一节是全文的核心每一层我都给出可直接复制的实现并且标注清楚输入输出边界。3.1 L1 模型层统一推理入口L1 的职责只有一个接收 prompt返回结构化响应。它不关心工具怎么执行不关心记忆怎么存。# l1_model.py import json from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, MODEL_REASONING, DEFAULT_TIMEOUT class ModelLayer: L1: 模型层 - 只负责 LLM 推理 def __init__(self, model_id: str MODEL_REASONING): self.client OpenAI( api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL, timeoutDEFAULT_TIMEOUT, ) self.model_id model_id def generate(self, prompt: str, tools: list None) - dict: 输入: prompt 字符串工具 schema 输出: {thought: str, action: str, action_input: dict} messages [{role: user, content: prompt}] kwargs { model: self.model_id, messages: messages, temperature: 0.2, } if tools: kwargs[tools] tools kwargs[tool_choice] auto resp self.client.chat.completions.create(**kwargs) msg resp.choices[0].message # 有工具调用 if msg.tool_calls: call msg.tool_calls[0] return { thought: msg.content or , action: call.function.name, action_input: json.loads(call.function.arguments), } # 纯文本回复 return { thought: msg.content or , action: finish, action_input: {}, }注意base_url指向 TaoToken 的 API 端点model_id从 config 读。这一层是唯一直接持有 client 的地方其他层通过它间接调模型。3.2 L2 能力层工具注册与执行L2 管工具的注册、schema 生成、执行。它不知道谁在调它只负责把工具跑起来。# l2_tools.py from typing import Callable class ToolLayer: L2: 能力层 - 工具注册与执行 def __init__(self): self._schemas {} self._handlers {} def register(self, name: str, description: str, parameters: dict, handler: Callable): self._schemas[name] { type: function, function: { name: name, description: description, parameters: parameters, }, } self._handlers[name] handler def get_schemas(self) - list: return list(self._schemas.values()) def execute(self, name: str, params: dict): if name not in self._handlers: raise ValueError(f工具 {name} 未注册) return self._handlers[name](**params) def get_weather(city: str) - str: return f{city}今天晴25°C def calculate(expr: str) - str: try: return str(eval(expr, {__builtins__: {}}, {})) except Exception as e: return f计算错误: {e}注册工具时parameters用 JSON Schema 格式这样能直接喂给 L1 的tools参数。3.3 L3 记忆层短期记忆摘要压缩L3 管上下文。短期记忆存最近 N 轮超阈值时调模型做摘要——注意这里用的是快档模型省成本。# l3_memory.py from openai import OpenAI from config import TAOTOKEN_API_KEY, TAOTOKEN_BASE_URL, MODEL_SUMMARY class MemoryLayer: L3: 记忆层 - 短期记忆与摘要 def __init__(self, max_entries: int 10): self.entries [] self.max_entries max_entries self.client OpenAI( api_keyTAOTOKEN_API_KEY, base_urlTAOTOKEN_BASE_URL, ) def add(self, role: str, content: str): self.entries.append({role: role, content: content}) if len(self.entries) self.max_entries: self._compress() def _compress(self): 超阈值时用快档模型压缩历史 text \n.join(f[{e[role]}] {e[content]} for e in self.entries) resp self.client.chat.completions.create( modelMODEL_SUMMARY, messages[{role: user, content: f用三句话总结以下对话的关键信息\n{text}}], ) summary resp.choices[0].message.content self.entries [{role: system, content: f[历史摘要] {summary}}] def get_context(self) - str: return \n.join(f[{e[role]}] {e[content]} for e in self.entries) def clear(self): self.entries []3.4 L4 编排层ReAct 循环L4 是大脑协调 L1、L2、L3。它持有三层的引用但只通过接口调用。# l4_orchestrator.py import json from l1_model import ModelLayer from l2_tools import ToolLayer from l3_memory import MemoryLayer class OrchestratorLayer: L4: 编排层 - ReAct 循环调度 def __init__(self, model: ModelLayer, tools: ToolLayer, memory: MemoryLayer, max_steps: int 5): self.model model self.tools tools self.memory memory self.max_steps max_steps def run(self, user_input: str) - str: self.memory.add(user, user_input) for step in range(self.max_steps): context self.memory.get_context() schemas self.tools.get_schemas() prompt self._build_prompt(context, schemas) resp self.model.generate(prompt, schemas) thought resp[thought] action resp[action] action_input resp[action_input] print(f[Step {step1}] Thought: {thought}) print(f[Step {step1}] Action: {action}({action_input})) self.memory.add(assistant, thought) if action finish: return thought try: obs self.tools.execute(action, action_input) except Exception as e: obs f执行错误: {e} print(f[Step {step1}] Observation: {obs}) self.memory.add(system, obs) return 达到最大步数限制 def _build_prompt(self, context: str, tools: list) - str: tool_desc json.dumps(tools, ensure_asciiFalse, indent2) return f你可以使用以下工具 {tool_desc} 历史记录 {context} 请根据用户需求决定下一步动作。如果需要调用工具直接调用如果任务完成直接回复最终答案。3.5 L5 交互层CLI 界面L5 只负责收输入、显示输出不碰任何业务逻辑。# l5_interface.py from l4_orchestrator import OrchestratorLayer class CLIInterface: L5: 交互层 - 命令行界面 def __init__(self, orchestrator: OrchestratorLayer): self.orchestrator orchestrator def start(self): print(Agent CLI 已启动输入 quit 退出) while True: user_input input(\n你: ).strip() if user_input.lower() in (quit, exit, q): break result self.orchestrator.run(user_input) print(f\nAgent: {result})3.6 组装入口# main.py from l1_model import ModelLayer from l2_tools import ToolLayer, get_weather, calculate from l3_memory import MemoryLayer from l4_orchestrator import OrchestratorLayer from l5_interface import CLIInterface def create_agent(): model ModelLayer() tools ToolLayer() tools.register( nameget_weather, description查询指定城市天气, parameters{ type: object, properties: {city: {type: string}}, required: [city], }, handlerget_weather, ) tools.register( namecalculate, description计算数学表达式, parameters{ type: object, properties: {expr: {type: string}}, required: [expr], }, handlercalculate, ) memory MemoryLayer(max_entries10) orchestrator OrchestratorLayer(model, tools, memory, max_steps5) return CLIInterface(orchestrator) if __name__ __main__: agent create_agent() agent.orchestrator.run(北京天气怎么样)到这里五层全部落地。每一层的输入输出边界都很清楚L1 进 prompt 出结构化响应L2 进工具名和参数出执行结果L3 进角色和内容出上下文字符串L4 进用户输入出最终答案L5 进键盘输入出屏幕输出。4. 端到端验证一次请求跑通五层链路代码写完了现在验证整条链路。运行main.py观察每一步的输出。export TAOTOKEN_API_KEYsk-your-key-here python main.py预期输出[Step 1] Thought: 用户想查北京天气我需要调用天气工具 [Step 1] Action: get_weather({city: 北京}) [Step 1] Observation: 北京今天晴25°C [Step 2] Thought: 已经拿到天气信息可以回复用户了 [Step 2] Action: finish({}) Agent: 北京今天晴25°C这条链路里L5 收到输入传给 L4L4 构建 prompt 调 L1L1 通过 TaoToken 通道请求模型返回工具调用L4 解析后调 L2 执行工具L2 返回结果L4 存入 L3再进下一轮循环最后 L5 显示结果。如果你想单独验证某一层可以写个最小测试# 只测 L1 from l1_model import ModelLayer m ModelLayer() print(m.generate(用一句话解释什么是 Agent)) # 只测 L2 from l2_tools import ToolLayer, get_weather t ToolLayer() t.register(get_weather, 查天气, {type: object, properties: {city: {type: string}}, required: [city]}, get_weather) print(t.execute(get_weather, {city: 上海}))分层的好处在这里体现得很明显哪层出问题就单独测哪层不用把整个 Agent 跑起来。验证模型对话是否正常可以用模型对话入口快速试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite5. 本篇常见错排查401、proxy failed、choices 为空这一节列几个你大概率会撞上的报错以及对应的排查路径。报错一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是 Key 没读到。检查config.py里的TAOTOKEN_API_KEY是否被环境变量覆盖成了空值。如果你在 shell 里export了但没生效用echo $TAOTOKEN_API_KEY确认。另一个常见原因是 Key 复制时带了空格或换行strip 一下。报错二local proxy failed / Connection erroropenai.APIConnectionError: Connection error.这类报错先确认base_url是不是写成了https://taotoken.net/api有没有多写斜杠或漏写/api。然后确认你的网络能正常访问该地址。如果你在容器里跑检查容器的 DNS 配置。报错三reading choices 为空IndexError: list index out of range出现在resp.choices[0]这一行。原因可能是模型返回了空响应或者你用的模型 ID 不存在。先打印完整的resp看看结构resp self.client.chat.completions.create(...) print(resp.model_dump())如果choices是空列表多半是模型 ID 写错了。回到 config 确认MODEL_REASONING的值和文档一致。报错四OAuth / 认证方式不匹配如果你在 Cline、Claude Code 这类工具里配置报 OAuth 相关错误说明工具默认走了 OAuth 流程而不是 API Key 流程。需要在工具设置里显式选择 API Key 模式然后填三件套Base URL:https://taotoken.net/apiAPI Key: 你创建的 KeyModel ID: 对应档位的模型三件套缺一不可。只填 Key 不填 Base URL工具会走默认端点只填 Base URL 不填 Model ID工具不知道调哪个模型。报错五工具调用参数解析失败json.decoder.JSONDecodeError: Expecting valueL1 里json.loads(call.function.arguments)这行挂了。原因是模型返回的 arguments 不是合法 JSON。加个容错try: args json.loads(call.function.arguments) except json.JSONDecodeError: args {}同时在 prompt 里强调参数必须是合法 JSON。排查完这些你的五层链路应该能稳定跑通。如果还有问题接入文档里有更详细的错误码说明https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite6. 把五层用起来下一步怎么走五层模型跑通之后你会发现加功能变得很轻松。想加一个新工具只在 L2 注册L1、L3、L4、L5 一行不改。想换个模型只改 L1 的model_id。想加个 Web 界面写个新的 L5复用同一个 orchestrator。如果你要长期做 Agent 开发建议把 Coding Plan 用起来它适合持续编码和 Agent 调试场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewriteAPI Key 管理在控制台https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite下一篇会聚焦 L1 模型层讲怎么根据任务类型选模型、怎么控制推理成本、怎么处理模型返回的不稳定结构。五层里 L1 是最容易被低估的一层选错模型会让整个 Agent 的表现打对折。