☰
AI Agent编程学习系列」第 2篇:Agent架构五层模型,从LLM内核到交互界面——用TaoToken统一Key跑通分层调用链
2026/10/9 23:22:15 网站建设 项目流程

1. 为什么你的 Agent 代码写到 500 行就崩了

我见过太多 Agent 项目死在同一个地方:一个agent.py文件里塞了模型调用、工具函数、对话历史、while 循环、还有 Flask 路由。刚开始跑得挺欢,加第三个工具的时候开始出 bug,加第五个工具的时候自己都不敢改了。

这不是你代码水平的问题,是架构缺了一张地图。Agent 系统本质上是个多层协作的软件工程问题,它至少包含五个职责完全不同的域:模型推理、工具执行、记忆存取、任务编排、用户交互。把这五件事揉在一起写,等于把数据库查询、HTTP 路由、业务逻辑全写进一个函数——能跑,但没法维护。

五层模型就是给 Agent 画一张分层地图:

层级名称核心职责典型技术
L1模型层LLM 推理与响应生成GPT、Claude、Qwen、DeepSeek
L2能力层工具定义、注册与执行Function Calling、MCP、API
L3记忆层短期与长期记忆管理上下文窗口、向量数据库
L4编排层任务规划、调度与执行控制ReAct、Plan-and-Execute
L5交互层用户界面与输入输出处理Web、CLI、Bot

分层之后你能得到什么?最直接的好处是替换成本骤降。想把 L1 从 GPT 换成 Claude,只改模型层的实现类;想给同一个 Agent 核心加个 Web 界面,只写一个新的 L5,L1-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 Key:

https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite

创建完 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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

现在把配置写成一个独立的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_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL, )

这里有个关键点:五层共用同一个 client 实例。不要每层各 new 一个,那样连接池、重试策略、超时配置都会分散,出问题时你没法统一调整。把 client 作为依赖注入到各层,这也是后面create_agent工厂函数要做的事。

如果你更习惯用 Claude Code 或 Cline 这类工具做开发辅助,它们的配置也是同一套逻辑——Base URL 填https://taotoken.net/api,Key 填你创建的那个,Model ID 填上面选的档位。三件套齐了就能跑。Coding Plan 适合长期编码场景,配置入口在这里:

https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

前置做完,你的项目结构应该是这样:

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_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL, timeout=DEFAULT_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_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_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( model=MODEL_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 {step+1}] Thought: {thought}") print(f"[Step {step+1}] 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 {step+1}] Observation: {obs}") self.memory.add("system", obs) return "达到最大步数限制" def _build_prompt(self, context: str, tools: list) -> str: tool_desc = json.dumps(tools, ensure_ascii=False, indent=2) 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( name="get_weather", description="查询指定城市天气", parameters={ "type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"], }, handler=get_weather, ) tools.register( name="calculate", description="计算数学表达式", parameters={ "type": "object", "properties": {"expr": {"type": "string"}}, "required": ["expr"], }, handler=calculate, ) memory = MemoryLayer(max_entries=10) orchestrator = OrchestratorLayer(model, tools, memory, max_steps=5) return CLIInterface(orchestrator) if __name__ == "__main__": agent = create_agent() agent.orchestrator.run("北京天气怎么样?")

到这里五层全部落地。每一层的输入输出边界都很清楚:L1 进 prompt 出结构化响应,L2 进工具名和参数出执行结果,L3 进角色和内容出上下文字符串,L4 进用户输入出最终答案,L5 进键盘输入出屏幕输出。

4. 端到端验证:一次请求跑通五层链路

代码写完了,现在验证整条链路。运行main.py,观察每一步的输出。

export TAOTOKEN_API_KEY="sk-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 收到输入传给 L4,L4 构建 prompt 调 L1,L1 通过 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_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite

5. 本篇常见错排查:401、proxy failed、choices 为空

这一节列几个你大概率会撞上的报错,以及对应的排查路径。

报错一:401 Unauthorized

openai.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 error

openai.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/api
  • API Key: 你创建的 Key
  • Model ID: 对应档位的模型

三件套缺一不可。只填 Key 不填 Base URL,工具会走默认端点;只填 Base URL 不填 Model ID,工具不知道调哪个模型。

报错五:工具调用参数解析失败

json.decoder.JSONDecodeError: Expecting value

L1 里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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

6. 把五层用起来:下一步怎么走

五层模型跑通之后,你会发现加功能变得很轻松。想加一个新工具?只在 L2 注册,L1、L3、L4、L5 一行不改。想换个模型?只改 L1 的model_id。想加个 Web 界面?写个新的 L5,复用同一个 orchestrator。

如果你要长期做 Agent 开发,建议把 Coding Plan 用起来,它适合持续编码和 Agent 调试场景:

https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

API Key 管理在控制台:

https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite

下一篇会聚焦 L1 模型层,讲怎么根据任务类型选模型、怎么控制推理成本、怎么处理模型返回的不稳定结构。五层里 L1 是最容易被低估的一层,选错模型会让整个 Agent 的表现打对折。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询