【万字长文】手写一个带记忆与 MCP 工具调用的 AI 智能体:从 ReAct 循环到 TaoToken 统一 Key 的完整落地

发布时间:2026/10/4 17:42:42
【万字长文】手写一个带记忆与 MCP 工具调用的 AI 智能体:从 ReAct 循环到 TaoToken 统一 Key 的完整落地 1. 从“套壳 API”到能干活AI 智能体到底缺了什么很多人第一次做 AI 应用写出来的东西长这样用户输入一句话拼进 Prompt调一次大模型接口把返回的字符串打印出来。跑通那一刻挺爽但用两天就发现不对劲——它不会查数据、不会算账、换个会话就把你忘得一干二净。这就是典型的“套壳 API”模型是别人的逻辑是死的能力上限就是一次问答。AI 智能体AI Agent要解决的就是这个上限问题。它让模型不再只输出一段文字而是能自己决定“下一步该干什么”需要算数就调计算器需要查库就调数据库干完一步看结果再决定下一步。支撑这套行为的核心机制叫 ReActReasoning Acting也就是“推理—行动”交替循环。而让 Agent 能接上外部世界的标准协议叫 MCPModel Context Protocol它把五花八门的工具接入统一成一种“插口”。再加上记忆MemoryAgent 才能跨会话记住你的偏好和历史决策。这篇文章适合谁适合已经会调大模型 API、但做出来的东西“只会聊天”的开发者适合想搞懂 LangGraph、CrewAI 这些框架底层到底在干什么的人也适合想给自己的小工具加一个“会自己动手”的智能层的人。我会用大约 300 行 Python从零手写一个带记忆、带 MCP 工具调用的智能体不依赖任何重量级框架每一步都给完整可复制的代码和配置。模型调用这一层我用 TaoToken 的统一 Key 和 API 通道来跑这样你不用在多个厂商的 Key 之间来回切换一个 Key 就能验证不同模型。先说清楚整体结构免得你写着写着迷路。我们要实现的东西分三层第一层是最小 ReAct 循环让模型学会“用工具而不是编答案”第二层加记忆解决跨会话失忆第三层接 MCP让工具生态可以无限扩展。三层是递进关系你可以先跑通第一层再往上加也可以直接照抄完整版。下面每一节我都会给出可运行的代码并且说明每个设计决策背后的原因——这些原因基本都是实际踩坑踩出来的不是教科书上的漂亮话。在动手之前先明确一个认知Agent 不是“更聪明的模型”而是“模型 循环 工具 记忆”的组合体。模型负责决策循环负责推进工具负责执行记忆负责积累。四者缺一不可。理解了这一点你再看任何 Agent 框架都能一眼看穿它在哪一层做了封装。2. 前置准备用 TaoToken 统一 Key 打通模型调用通道在写 Agent 之前得先解决模型调用这一层。传统做法是每个厂商注册一个账号、拿一个 Key、记一套 Base URL切换模型时改代码改配置非常烦。我这次用 TaoToken 来做统一通道它的思路是提供一个 OpenAI 兼容的接口你用同一个 Key 就能调用不同模型Agent 代码里只需要改一个模型名字符串。TaoToken 的官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。注意 API 地址后面不加任何参数直接作为 OpenAI 客户端的 base_url 使用即可。它的接口是 OpenAI 兼容格式这意味着我们后面写的所有代码用的都是标准的openaiPython SDK不需要任何私有 SDK。具体怎么拿 Key进入控制台后创建 API Key复制出来保存好。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API Keys 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。拿到 Key 之后先别急着写 Agent用一段最小代码验证通道是否通。这一步很重要因为后面 Agent 出问题时你要能快速判断是“模型通道不通”还是“Agent 逻辑有 bug”。环境准备方面建议 Python 3.10 以上3.11 更稳。依赖装这几个python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install openai1.55.0 python-dotenv1.0.1 chromadb0.5.23 mcp1.2.0其中openai是调用 SDKpython-dotenv用来读环境变量chromadb用于长期记忆的向量检索mcp是可选的 MCP 客户端库——不装也能跑通前两层只是少了 MCP 演示。如果你只想先跑最小版可以暂时只装前两个。配置用.env文件管理不要硬编码 Key# .env OPENAI_BASE_URLhttps://taotoken.net/api OPENAI_API_KEYsk-你从控制台复制的key OPENAI_MODEL你选用的模型ID这里OPENAI_BASE_URL填 TaoToken 的 API 地址OPENAI_API_KEY填你创建的 KeyOPENAI_MODEL填你想用的模型 ID。因为 TaoToken 是统一通道你换模型只需要改这一行代码完全不用动。这一点在 Agent 开发里特别有用——不同模型对结构化输出的遵循程度不一样你可能需要试几个模型才能找到最稳的那个统一通道让这个试错成本变得极低。验证通道的最小代码# check_channel.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) resp client.chat.completions.create( modelos.getenv(OPENAI_MODEL), messages[{role: user, content: 只回复两个字通了}], temperature0, ) print(resp.choices[0].message.content)跑一下python check_channel.py如果输出“通了”说明通道没问题可以进入下一步。如果报错先看第 5 节的排错部分那里列了几个最常见的错误和对应原因。这一步别跳过我见过太多人一上来就写几百行 Agent结果卡在 Key 配错上白白浪费时间。3. 可复制配置智能体骨架与工具注册表这一节给出智能体的骨架配置包括项目结构、工具注册表、以及第一层 ReAct 循环的完整代码。你可以直接复制到本地跑。项目结构建议这样组织清晰且方便后续扩展agent_lab/ ├── agent.py # 核心智能体与 ReAct 循环 ├── tools.py # 工具注册表与内置工具 ├── memory.py # 两级记忆实现 ├── mcp_tools.py # MCP 客户端封装可选 ├── main.py # 交互入口 ├── .env └── requirements.txt先写工具注册表tools.py。核心思路是用一个装饰器把函数注册进全局字典Agent 只需要知道工具名和参数说明就能调用。这种设计让新增工具变成一行代码的事# tools.py import json TOOL_REGISTRY {} def tool(name): def decorator(fn): TOOL_REGISTRY[name] fn return fn return decorator tool(calculator) def calculator(expression): 安全计算只允许数字和四则运算符。 safe expression.replace( , ) if not all(c.isdigit() or c in -*/.() for c in safe): return json.dumps({error: 非法表达式只支持数字和 -*/()}) try: return json.dumps({result: eval(safe, {__builtins__: {}}, {})}) except Exception as e: return json.dumps({error: str(e)}) tool(get_weather) def get_weather(city): 演示工具返回模拟天气数据生产环境替换为真实 API。 mock {北京: 晴 32°C, 上海: 多云 29°C, 广州: 雷阵雨 28°C} return json.dumps({city: city, weather: mock.get(city, 暂无数据)}) def execute_tool(name, action_input): if name not in TOOL_REGISTRY: return json.dumps({error: f未知工具: {name}}) try: args json.loads(action_input) if action_input else {} return TOOL_REGISTRY[name](**args) except TypeError as e: return json.dumps({error: f参数错误: {e}})注意calculator里用了eval但做了字符白名单过滤只允许数字和四则运算符。这是教学演示的降级方案生产环境请用 AST 解析或asteval绝不对任意输入执行eval。这个坑我在第 5 节还会再强调一次。接下来是核心的agent.py包含系统提示词和 ReAct 循环。系统提示词是整个 Agent 的灵魂它规定了模型必须以 JSON 格式输出决策包含thought、action、action_input、done四个字段# agent.py import json import os from openai import OpenAI from dotenv import load_dotenv from tools import execute_tool load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) MODEL os.getenv(OPENAI_MODEL) SYSTEM_PROMPT 你是一个能调用工具的智能体。 所有回复必须是一个 JSON 对象包含四个字段 { thought: 你的推理过程说明为什么这么做, action: 要调用的工具名不需要工具时填 FINISH, action_input: 传给工具的参数JSON 字符串, done: true 或 false } 可用工具及参数说明 - calculator(expression): 计算数学表达式如 1 2 * 3 - get_weather(city): 查询城市天气 推理规则 1. 需要算数/查天气时必须调用工具绝不编造结果 2. 拿到工具结果后再决定是继续调用还是输出最终答案 3. 最终答案放在 thought 字段中action 置 FINISHdone 置 true def call_llm(messages): resp client.chat.completions.create( modelMODEL, messagesmessages, temperature0.2, ) return resp.choices[0].message.content def parse_decision(raw): text raw.strip() if text.startswith(): text text.split(\n, 1)[1].rsplit(, 1)[0].strip() try: return json.loads(text) except json.JSONDecodeError: return {thought: text, action: FINISH, action_input: {}, done: True} def run_agent(user_query, max_steps8): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_query}, ] for step in range(max_steps): raw call_llm(messages) decision parse_decision(raw) action decision.get(action) if action FINISH or decision.get(done): print(f\n[最终答案] {decision[thought]}) return decision[thought] result execute_tool(action, decision.get(action_input, {})) print(f[第{step1}步] 调用 {action} - {result}) messages.append({role: assistant, content: raw}) messages.append({role: user, content: f工具返回结果: {result}。请继续推理。}) return 已达最大步数请确认任务是否完成。这里有几个关键设计点值得说明。第一max_steps上限必须设否则模型可能陷入工具调用死循环一次调用烧掉大量 Token。第二parse_decision做了容错模型有时会用 json 包裹输出有时干脆输出一段自然语言兜底逻辑把整段话当最终答案避免程序崩溃。第三每轮把模型的原始输出和工具结果都追加进messages让模型“看到”自己上一步干了什么这是 ReAct 循环能推进的前提。如果你用的是 Claude Code 这类工具做辅助开发配置方式类似核心三件套是 Base URL、Key、Model ID。Base URL 填https://taotoken.net/apiKey 填你的 TaoToken KeyModel ID 填你选的模型。这三样配对了通道就通了。Cline、Codex 的auth.json配置逻辑也一样都是这三个字段只是文件位置和字段名略有差异。4. 验证请求跑通端到端并观察成功结果配置写完了现在跑起来验证。先写一个简单的交互入口main.py# main.py from agent import run_agent if __name__ __main__: print(AI 智能体已启动输入 exit 退出) while True: query input(\n你: ) if query.strip().lower() exit: break run_agent(query)运行python main.py然后输入一个需要算数的复合问题比如“营业额 12893.5 元成本率 38%利润是多少”。你会看到类似这样的输出你: 营业额 12893.5 元成本率 38%利润是多少 [第1步] 调用 calculator - {result: 4899.53} [第2步] 调用 calculator - {result: 7993.97} [最终答案] 营业额 12893.5 元成本率 38%成本为 4899.53 元利润约为 7993.97 元。看到这个结果说明 ReAct 循环跑通了模型先推理出需要算成本调用计算器拿到结果后再算利润最后输出答案。整个过程模型没有编造数字而是真的调用了工具。这就是 Agent 和“套壳 API”的本质区别。再试一个天气查询“明天去上海出差帮我看看天气。”输出应该是你: 明天去上海出差帮我看看天气。 [第1步] 调用 get_weather - {city: 上海, weather: 多云 29°C} [最终答案] 上海明天多云29°C建议带薄外套。到这里第一层最小 ReAct 循环就验证通过了。接下来加记忆。记忆分两级短期记忆就是对话窗口里直接带着的上下文第一层已经做到了长期记忆用向量库存储跨会话可检索。memory.py的实现如下# memory.py import sqlite3 import hashlib from datetime import datetime class MemoryManager: def __init__(self, db_pathagent_memory.db, use_vectorTrue): self.use_vector use_vector self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS facts( id INTEGER PRIMARY KEY AUTOINCREMENT, key TEXT UNIQUE, value TEXT, updated_at TEXT ) ) self.collection None if use_vector: try: import chromadb _client chromadb.Client() self.collection _client.get_or_create_collection(agent_memory) except Exception: self.use_vector False def remember(self, key, value): self.conn.execute( INSERT OR REPLACE INTO facts(key, value, updated_at) VALUES(?,?,?), (key, value, datetime.now().isoformat()), ) self.conn.commit() def recall_fact(self, key): row self.conn.execute(SELECT value FROM facts WHERE key?, (key,)).fetchone() return row[0] if row else None def store_semantic(self, text, metaNone): if not self.use_vector: return _id hashlib.md5(text.encode()).hexdigest()[:16] self.collection.upsert( ids[_id], documents[text], metadatas[meta or {time: datetime.now().isoformat()}], ) def search_semantic(self, query, top_k3): if not self.use_vector or self.collection.count() 0: return [] res self.collection.query(query_texts[query], n_resultsmin(top_k, self.collection.count())) return res[documents][0] if res.get(documents) else []把记忆接进 Agent改造入口函数# agent.py 追加 from memory import MemoryManager memory MemoryManager() def run_agent_with_memory(user_query, user_iddefault): history memory.search_semantic(user_query, top_k3) pref memory.recall_fact(fuser:{user_id}:preference) context_prompt if history: context_prompt 以下是与本次问题相关的历史经验供参考\n \n---\n.join(history) \n if pref: context_prompt f该用户已知偏好{pref}\n result run_agent(context_prompt user_query) memory.store_semantic(f用户({user_id})问{user_query}回答{result}) return result验证记忆是否生效先问一次“营业额 12893.5 元成本率 38%利润是多少”退出程序重新启动再问“还记得我上次关心的利润算法吗”。如果 Agent 能检索到上次的对话并回答出计算方法说明长期记忆生效了。这里的关键是“检索式记忆”——不是把全部历史塞给模型而是只取与当前问题最相关的几段这样既不爆上下文窗口又能跨会话记住关键信息。最后是 MCP 接入。mcp_tools.py封装一个 stdio 方式的 MCP 客户端# mcp_tools.py import asyncio import json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def query_mcp_server(server_cmd, server_args, tool_name, arguments): server_params StdioServerParameters(commandserver_cmd, argsserver_args) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() names [t.name for t in tools.tools] if tool_name not in names: return json.dumps({error: fMCP Server 无此工具可用: {names}}) result await session.call_tool(tool_name, json.loads(arguments)) return json.dumps({mcp_result: [c.text for c in result.content]}) def mcp_sqlite_call(sql): return asyncio.run(query_mcp_server( server_cmduvx, server_args[mcp-server-sqlite, --db-path, ./demo.db], tool_nameread_query, argumentsjson.dumps({query: sql}, ensure_asciiFalse), ))然后在tools.py里注册try: from mcp_tools import mcp_sqlite_call tool(mcp_sqlite_query) def mcp_sqlite_query(query): 通过 MCP 协议查询本地 SQLite 数据库。query 为 SQL 语句。 return mcp_sqlite_call(query) print([OK] MCP 工具已启用) except ImportError: print([跳过] 未安装 mcp 库MCP 工具不可用)对 Agent 来说mcp_sqlite_query只是一个普通工具名它不需要知道底层是 SQLite 还是别的什么。这就是 MCP 的价值把工具接入从 N×M 的适配地狱变成一次对接、处处可用。验证方式是问“用数据库查一下 demo.db 里 employees 表有多少人”如果返回记录数说明 MCP 链路通了。5. 本篇常见错排查401、local proxy failed 与 choices 解析跑 Agent 的过程中报错是常态。这一节列几个我实际遇到过的典型错误以及对应的排查思路。这些错误覆盖了从通道配置到代码逻辑的各个环节你按顺序排查基本能定位问题。第一个高频错误是 401 认证失败。典型报错长这样openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key, ...}}原因通常是 Key 复制时带了空格、换行或者.env文件里 Key 写错了变量名。排查步骤先确认.env里OPENAI_API_KEY的值没有多余字符可以用print(repr(os.getenv(OPENAI_API_KEY)))打印出来看再确认OPENAI_BASE_URL填的是https://taotoken.net/api注意结尾不要多加/v1或斜杠不同通道对路径的处理不一样。如果 Key 本身没问题去控制台确认这个 Key 是否被禁用或额度耗尽。第二个常见错误是 local proxy failed 或连接超时。报错类似openai.APIConnectionError: Connection error.这种一般是网络层的问题。先确认你的机器能正常访问外网再确认OPENAI_BASE_URL没有写错。如果你在公司内网可能有防火墙限制需要走公司允许的出口。注意这里不要尝试任何绕过网络管理的手段合规使用是前提。如果确认网络正常检查是不是base_url末尾多了斜杠导致路径拼接错误比如https://taotoken.net/api/和https://taotoken.net/api在某些 SDK 版本下行为不同建议去掉末尾斜杠。第三个错误是解析choices时出错典型报错IndexError: list index out of range或者AttributeError: NoneType object has no attribute choices这通常发生在resp.choices[0]这一行。原因可能是接口返回了错误结构但 SDK 没抛异常导致choices为空。排查方法在call_llm里加一层打印把原始响应打出来看resp client.chat.completions.create(...) print(resp) # 调试时打开 return resp.choices[0].message.content如果resp里没有choices说明请求本身有问题回到 401 或连接错误的排查。如果choices有值但内容为空可能是模型返回了空字符串检查你的 Prompt 是否让模型困惑。第四个错误和 OAuth 或认证方式有关。如果你用的是 Claude Code、Cline 这类工具配置 MCP 或模型通道时可能遇到 OAuth 相关的报错。核心还是那三件套Base URL、Key、Model ID。以 Claude Code 为例配置里需要明确指定 API 端点和认证方式Base URL 填https://taotoken.net/apiKey 填 TaoToken 的 KeyModel ID 填你选的模型。如果工具提示 OAuth 失败先确认是不是把 API Key 认证和 OAuth 认证搞混了——这两套机制不一样API Key 方式不需要走 OAuth 流程。第五个错误是 MCP Server 启动失败。典型报错FileNotFoundError: [Errno 2] No such file or directory: uvx这是因为uvx命令没装。uvx是 uv 工具链的一部分需要先安装 uv。另一个常见问题是 MCP Server 的--db-path指向的数据库文件不存在SQLite 不会自动创建需要你先手动建库建表。排查时先用命令行单独跑一下 MCP Server确认它能正常启动再接到 Agent 里。第六个错误是工具调用死循环。表现是 Agent 反复调用同一个工具步数用尽还没结束。原因通常是工具返回的结果模型看不懂或者 Prompt 里没告诉模型“拿到结果后该怎么办”。解决办法是在工具返回里加上明确的字段说明并在系统提示词里强调“拿到工具结果后如果信息足够就输出最终答案”。另外max_steps一定要设这是最后一道防线。把这几类错误过一遍基本能覆盖 90% 的入门问题。剩下的就是模型本身的“脾气”——不同模型对 JSON 格式的遵循程度不一样有的模型经常输出多余的解释文字。遇到这种情况可以在 Prompt 里加一句“只输出 JSON不要任何其他文字”或者在parse_decision里做更强的容错提取。6. 语义一致 CTA把统一 Key 用进你的长期编码流跑通这个 Agent 之后你会发现模型调用这一层其实是最不该操心的部分。真正花时间的是 ReAct 循环的稳定性、记忆的检索质量、MCP 工具的接入调试。所以把模型通道统一起来用 TaoToken 一个 Key 管所有模型能省下大量切换配置的时间。如果你主要在做这种长期编码、Agent 开发的事情建议直接上 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它适合需要频繁调用模型、反复试不同模型效果的场景比按次调用更划算。日常想快速验证某个模型对结构化输出的遵循程度可以用模型对话页面直接试地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 不用写代码就能对比不同模型的表现。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面写了各种语言和工具的接入方式包括 Claude Code、Cline、Codex 这些常用工具的配置示例。API Keys 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 控制台在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。回到 Agent 本身最后给你一个实用建议把工具调用的入参、出参、耗时都记下来。我试过在排查“模型为什么乱调工具”时全靠这份日志定位问题。你可以在execute_tool里加一行日志记录工具名、参数、返回值和耗时存到本地文件或 SQLite。这份日志在调试阶段的价值比任何花哨的框架都高。Agent 的可观测性就是从这一行日志开始的。