☰
【万字长文】手写一个带记忆与 MCP 工具调用的 AI 智能体:从 ReAct 循环到 TaoToken 统一 Key 的完整落地
2026/10/4 17:42:30 网站建设 项目流程

1. 从“套壳 API”到能干活:AI 智能体到底缺了什么

很多人第一次做 AI 应用,写出来的东西长这样:用户输入一句话,拼进 Prompt,调一次大模型接口,把返回的字符串打印出来。跑通那一刻挺爽,但用两天就发现不对劲——它不会查数据、不会算账、换个会话就把你忘得一干二净。这就是典型的“套壳 API”:模型是别人的,逻辑是死的,能力上限就是一次问答。

AI 智能体(AI Agent)要解决的就是这个上限问题。它让模型不再只输出一段文字,而是能自己决定“下一步该干什么”,需要算数就调计算器,需要查库就调数据库,干完一步看结果再决定下一步。支撑这套行为的核心机制叫 ReAct(Reasoning + Acting),也就是“推理—行动”交替循环。而让 Agent 能接上外部世界的标准协议叫 MCP(Model Context Protocol),它把五花八门的工具接入统一成一种“插口”。再加上记忆(Memory),Agent 才能跨会话记住你的偏好和历史决策。

这篇文章适合谁?适合已经会调大模型 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api 。注意 API 地址后面不加任何参数,直接作为 OpenAI 客户端的 base_url 使用即可。它的接口是 OpenAI 兼容格式,这意味着我们后面写的所有代码,用的都是标准的openaiPython SDK,不需要任何私有 SDK。

具体怎么拿 Key:进入控制台后创建 API Key,复制出来保存好。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API Keys 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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 openai==1.55.0 python-dotenv==1.0.1 chromadb==0.5.23 mcp==1.2.0

其中openai是调用 SDK,python-dotenv用来读环境变量,chromadb用于长期记忆的向量检索,mcp是可选的 MCP 客户端库——不装也能跑通前两层,只是少了 MCP 演示。如果你只想先跑最小版,可以暂时只装前两个。

配置用.env文件管理,不要硬编码 Key:

# .env OPENAI_BASE_URL=https://taotoken.net/api OPENAI_API_KEY=sk-你从控制台复制的key OPENAI_MODEL=你选用的模型ID

这里OPENAI_BASE_URL填 TaoToken 的 API 地址,OPENAI_API_KEY填你创建的 Key,OPENAI_MODEL填你想用的模型 ID。因为 TaoToken 是统一通道,你换模型只需要改这一行,代码完全不用动。这一点在 Agent 开发里特别有用——不同模型对结构化输出的遵循程度不一样,你可能需要试几个模型才能找到最稳的那个,统一通道让这个试错成本变得极低。

验证通道的最小代码:

# check_channel.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"), ) resp = client.chat.completions.create( model=os.getenv("OPENAI_MODEL"), messages=[{"role": "user", "content": "只回复两个字:通了"}], temperature=0, ) 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_key=os.getenv("OPENAI_API_KEY"), base_url=os.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 置 'FINISH',done 置 true """ def call_llm(messages): resp = client.chat.completions.create( model=MODEL, messages=messages, temperature=0.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_steps=8): 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"[第{step+1}步] 调用 {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/api,Key 填你的 TaoToken Key,Model 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_path="agent_memory.db", use_vector=True): 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, meta=None): 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_k=3): if not self.use_vector or self.collection.count() == 0: return [] res = self.collection.query(query_texts=[query], n_results=min(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_id="default"): history = memory.search_semantic(user_query, top_k=3) pref = memory.recall_fact(f"user:{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(command=server_cmd, args=server_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": f"MCP 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_cmd="uvx", server_args=["mcp-server-sqlite", "--db-path", "./demo.db"], tool_name="read_query", arguments=json.dumps({"query": sql}, ensure_ascii=False), ))

然后在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/api,Key 填 TaoToken 的 Key,Model 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。它适合需要频繁调用模型、反复试不同模型效果的场景,比按次调用更划算。日常想快速验证某个模型对结构化输出的遵循程度,可以用模型对话页面直接试,地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,不用写代码就能对比不同模型的表现。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,里面写了各种语言和工具的接入方式,包括 Claude Code、Cline、Codex 这些常用工具的配置示例。API Keys 管理在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,控制台在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。

回到 Agent 本身,最后给你一个实用建议:把工具调用的入参、出参、耗时都记下来。我试过在排查“模型为什么乱调工具”时,全靠这份日志定位问题。你可以在execute_tool里加一行日志,记录工具名、参数、返回值和耗时,存到本地文件或 SQLite。这份日志在调试阶段的价值,比任何花哨的框架都高。Agent 的可观测性,就是从这一行日志开始的。

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

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

立即咨询