不止会用框架,自己动手写一个带记忆和工具的智能体,真的没那么难。
前言
我写过不少 Agent 应用,也用过 OpenClaw、LangChain 这类框架。但说真的,只有自己从头写一遍,才真正理解 Agent 是怎么工作的。
今天这篇文章,我把最近写的一个 Python Agent 完整分享出来。它具备:
长期记忆(基于 MySQL,支持多会话)
工具调用(天气查询、数学计算,可无限扩展)
ReAct 风格的自主推理
命令行交互 + 会话管理
所有代码都可以直接复制运行,我会把每一步的坑都标出来。
一、为什么要自己写,而不是直接用框架?
市面上 Agent 框架很多,但新手往往会遇到几个问题:
配置复杂,依赖冲突
黑盒太多,不知道内部到底发生了什么
调试困难,出错不知道改哪里
Token 消耗巨大(框架经常调用多次 LLM)
自己写一个简化版 Agent,代码不超过 200 行,却能让你彻底理解 记忆存储、工具调用、消息循环 这三个核心机制。
二、整体架构设计
我们的 Agent 包含五个模块:
text
1. 数据库层 → 存储所有会话消息(JSON格式)
2. LLM 层 → 调用阿里云 DashScope(兼容OpenAI接口)
3. 工具层 → 定义函数 + 描述(JSON Schema)
4. 执行器 → 根据 LLM 返回的 tool_calls 调用真实函数
5. 交互层 → 命令行 + 会话切换命令
流程图:
三、环境准备
3.1 安装依赖
bash
pip install pymysql openai
注意:openai 库需要 1.0 以上版本,建议 pip install --upgrade openai
3.2 安装 MySQL(如果没有)
启动 MySQL,创建一个数据库:
sql
CREATEDATABASE agent_db;
3.3 获取 LLM API Key
我使用的是 阿里云百炼平台(qwen-plus 模型),新用户有免费额度。也可以换成 DeepSeek、OpenAI 等,只需要修改 base_url 和 api_key。
四、完整代码(逐段讲解)
4.1 数据库配置和初始化
python
DB_CONFIG ={"host":"localhost","user":"root","password":"你的MySQL密码",# 务必修改"database":"agent_db","port":3306,"charset":"utf8mb4"}初始化表结构(自动创建 conversation_messages 表):
# ================= 初始化 MySQL 表结构 =================def init_db(): """确保 conversation_messages 表存在""" conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS conversation_messages ( id INT AUTO_INCREMENT PRIMARY KEY, session_id VARCHAR(64) NOT NULL, message_data JSON NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_session_id (session_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) conn.commit() cursor.close() conn.close()
4.2 消息存取函数
python
# ================= 保存单条消息到数据库 =================def save_message(session_id: str, message_dict: dict): """将一条消息(dict格式)存入数据库""" conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor() json_str = json.dumps(message_dict, ensure_ascii=False) cursor.execute( "INSERT INTO conversation_messages (session_id, message_data) VALUES (%s, %s)", (session_id, json_str) ) conn.commit() cursor.close() conn.close()# ================= 加载指定会话的所有历史消息 =================def load_history(session_id: str): """返回该会话的消息列表(按时间升序)""" conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute( "SELECT message_data FROM conversation_messages WHERE session_id = %s ORDER BY id ASC", (session_id,) ) rows = cursor.fetchall() cursor.close() conn.close() messages = [json.loads(row[0]) for row in rows] return messages# ================= 获取所有已存在的会话ID =================def list_sessions(): """返回数据库中的所有 session_id(去重)""" conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute("SELECT DISTINCT session_id FROM conversation_messages ORDER BY session_id") sessions = [row[0] for row in cursor.fetchall()] cursor.close() conn.close() return sessions
L
4.3 LLM 客户端(使用阿里云 DashScope)
# ================= DeepSeek 客户端配置 =================client = OpenAI( api_key="sk-b0fa1ae8668449a4815b01d268419758", # 请替换为你的真实 API Key base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")
4.4 定义工具函数
#================= 工具函数(天气、计算) =================def get_weather(city: str) -> str: weather_db = {"北京": "晴天,30度", "上海": "小雨,22度", "广州": "雷阵雨,30度"} return weather_db.get(city, f"未找到{city}的天气")def calculate(expression: str) -> str: try: result = eval(expression) return f"计算结果: {result}" except: return "表达式错误"
4.5 工具描述(给 LLM 看的)
python
tools = [ { "type": "function", "function": { "name": "get_weather", "description": "查询某个城市的天气", "parameters": { "type": "object", "properties": {"city": {"type": "string", "description": "城市名"}}, "required": ["city"] } } }, { "type": "function", "function": { "name": "calculate", "description": "计算数学表达式", "parameters": { "type": "object", "properties": {"expression": {"type": "string", "description": "数学表达式"}}, "required": ["expression"] } } }]
4.6 Agent 核心执行器
python
available_functions = {"get_weather": get_weather, "calculate": calculate}def run_agent(messages, session_id): """ 处理当前对话历史(内存中的 messages 列表), 调用模型、执行工具、保存新消息到数据库,并返回更新后的 messages。 """ response = client.chat.completions.create( model="qwen-plus", messages=messages, tools=tools, tool_choice="auto" ) response_message = response.choices[0].message tool_calls = response_message.tool_calls # 保存模型的初始响应 msg_dict = response_message.model_dump() messages.append(msg_dict) save_message(session_id, msg_dict) if not tool_calls: print("Agent:", response_message.content) return messages # 执行工具调用 for tool_call in tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments) result = available_functions[func_name](**func_args) print(f"[工具调用] {func_name}({func_args}) -> {result}") tool_msg = { "role": "tool", "tool_call_id": tool_call.id, "content": result } messages.append(tool_msg) save_message(session_id, tool_msg) # 生成最终回复 final_response = client.chat.completions.create( model="qwen-plus", messages=messages ) final_message = final_response.choices[0].message final_dict = final_message.model_dump() messages.append(final_dict) save_message(session_id, final_dict) print("Agent:", final_message.content) return messages
4.7 交互式命令行(支持会话管理)
实现了以下命令:
/new – 新建会话
/list – 列出所有历史会话
/switch <会话名> – 切换到已有会话
/exit – 退出
五、运行效果演示
启动程序:
bash
python agent.py
首次运行会提示新建会话,输入 test 回车。
text
你: 北京今天天气怎么样?[工具调用] get_weather({'city': '北京'}) -> 晴天,30度Agent: 北京今天天气晴朗,气温30度。text
你: 计算 123 * 456[工具调用] calculate({'expression': '123*456'}) -> 计算结果: 56088Agent: 123乘以456的结果是56088。切换会话:
text
你: /new请输入新会话名称: 数学专聊已切换到新会话: 数学专聊,历史为空。
你会发现切换到另一个会话后,之前的对话完全独立,互不干扰。
六、你可以扩展的功能
增加更多工具
联网搜索(调用 SerpAPI 或自定义爬虫)
发送邮件(smtplib)
读写本地文件(开放一定权限)
执行 Shell 命令(注意安全!)
接入更多渠道
微信(使用 itchat 或企业微信 API)
钉钉、飞书机器人
Telegram Bot
优化记忆
限制会话长度(自动裁剪历史)
添加向量数据库(RAG)支持长文本检索
支持流式输出
七、避坑指南
7.1 关于 API Key
代码中直接写了 api_key,发布到公众号时我已经打码。你自己使用时务必替换成真实的 Key,并且不要提交到公开 GitHub 仓库。
7.2 数据库连接失败
7.3 工具调用不触发
检查 tools 中的 description 是否清晰
确认 tool_choice="auto"(不要强制某个工具)
尝试问:“调用 get_weather 查询北京天气” 这种明确指令
7.4 Token 消耗
自己写的 Agent 比 OpenClaw 这类框架省很多 Token。因为:
只有需要工具时才二次调用
没有额外的系统提示词和插件开销
实测一次简单对话消耗不到 200 tokens。
八、总结
通过这 200 行代码,你实现了一个具备以下能力的 Agent:
| |
|---|
| |
| |
| OpenAI Function Calling 标准 |
| |
框架会过时,但底层原理不会。
如果你想深入学习 Agent 开发,强烈建议自己从头写一个最小版本。当你把这段代码跑通的那一刻,你会对 LangChain、OpenClaw 等框架的内部机制豁然开朗。
附录:完整代码获取
由于公众号篇幅限制,完整代码请关注本公众号回复‘agent代码’获取,或者联系小夭也可以,或者直接复制文章中分散的代码块,按顺序拼接即可运行。
如果你在运行中遇到任何问题,欢迎留言,我会尽力解答。
如果觉得这篇文章对你有帮助,点赞、在看、转发 支持一下,让更多同学看到从零写 Agent 并不难。