129 行 Python，手写一个 AI 智能体

以下包含代码及注释仅在倒数第四行配置自己的deepseek-key即可运行mini_agent.py — 从零实现一个 AI 智能体基于 ReAct (Thought → Action → Observation) 范式核心概念:Agent 大脑(LLM) 工具(Tools) 记忆(Messages) 循环(Loop)运行流程:用户输入 → [思考(Thought) → 行动(Action) → 观察(Observation)] × N → 最终答案运行前提: pip install openai requests使用方法: python mini_agent.pyimport re, requestsfrom openai import OpenAI# ═══════════════════════════════════════════════════════════════# 1. 工具系统# 工具 智能体的手和眼睛让它能获取外部信息、执行操作。# tool 装饰器将普通函数注册到 TOOL_REGISTRY# LLM 通过函数名称 参数的方式调用它们。# ═══════════════════════════════════════════════════════════════TOOL_REGISTRY {}def tool(name, desc):def deco(func):TOOL_REGISTRY[name] {func: func, desc: desc}return funcreturn decotool(calculator, 数学计算器例如: calculator(expr1 2 * 3))def calculator(expr: str) - str:把 LLM 给的数学表达式 eval 算出结果expr_clean expr.replace(×, *).replace(÷, /).replace(x, *)try:return f计算结果: {eval(expr_clean)}except Exception as e:return f计算错误: {e}tool(get_time, 获取指定城市的当前时间例如: get_time(city北京))def get_time(city: str) - str:import datetimereturn f{city}当前时间: {datetime.datetime.now():%Y-%m-%d %H:%M:%S}tool(get_weather, 查询指定城市的实时天气例如: get_weather(city北京))def get_weather(city: str) - str:通过免费 API wttr.in 获取实时天气try:data requests.get(fhttps://wttr.in/{city}?formatj1, timeout10).json()c data[current_condition][0]return f{city}天气: {c[weatherDesc][0][value]}气温{c[temp_C]}°Cexcept Exception as e:return f天气查询失败: {e}# ═══════════════════════════════════════════════════════════════# 2. Agent 智能体# 集大脑(LLM)、记忆(messages)、循环(run)于一身。# 不需要单独的 LLMClient 类——Agent 自己管理 API 调用。# ═══════════════════════════════════════════════════════════════class Agent:def __init__(self, model: str, api_key: str, base_url: str):# LLM 客户端 — 智能体的大脑负责推理决策self.llm OpenAI(api_keyapi_key, base_urlbase_url)self.model model# 对话记忆 — 存储完整的思考→行动→观察历史self.messages []def _build_prompt(self) - str:构造系统提示词告诉 LLM 它的身份、工具和输出格式tools_text \n.join(f - {name}: {info[desc]}for name, info in TOOL_REGISTRY.items())return (你是一个通用的 AI 智能体。你可以使用工具来获取信息或执行操作。\nf\n【可用工具】\n{tools_text}\n\n【工作方式】\n请通过思考→行动循环来解决问题每轮只输出一对内容\n Thought: [你的思考过程]\n Action: [工具名称(参数名参数值)]\n\n当你认为已经完成任务时使用以下格式结束\n Thought: [总结]\n Action: Finish[最终答案]\n\n【规则】\n- 一次只输出一对 Thought 和 Action\n- Action 必须在一行内开始参数用双引号)def _call_llm(self) - str:调用大语言模型返回回复内容或错误信息print( 思考中..., end, flushTrue)try:resp self.llm.chat.completions.create(modelself.model, messagesself.messages, temperature0.3, timeout30)print(\r LLM 响应完成)return resp.choices[0].message.contentexcept Exception as e:print(f\r❌ LLM 调用失败: {type(e).__name__})return f错误: {type(e).__name__}: {e}def _parse_action(self, response: str):从 LLM 的回复中解析 Action 字段m re.search(rAction:\s*(.*), response)if not m:return {error: 回复中缺少 Action 字段}action m.group(1).strip()# 检查是否结束if action.startswith(Finish[):if action.endswith(]):return {type: finish, answer: action[7:-1]}# 支持多行 Finish[...\n内容\n] 的情况fm re.search(rFinish\[(.*?)\], response, re.DOTALL)return {type: finish, answer: fm.group(1).strip() if fm else action[7:]}# 解析工具调用m re.match(r(\w)\((.*)\), action)if not m:return {error: f无法解析行动格式: {action}}args dict(re.findall(r(\w)[\]([^\]*)[\], m.group(2)))return {type: tool, name: m.group(1), args: args}def run(self, user_input: str, max_steps: int 10):主循环感知→思考→行动→观察直到得出最终答案self.messages [{role: system, content: self._build_prompt()},{role: user, content: user_input}]print(f\n{*50}\n 用户: {user_input}\n{*50})for step in range(max_steps):print(f\n--- 第 {step1} 轮循环 ---)# 【思考】调用 LLM 进行推理决策response self._call_llm()if not response or response.startswith(错误:):print(f\n❌ {response or LLM 返回为空})print( 提示: 请检查 API Key、base_url 和 model 名称是否正确)breakprint(f\n{response}\n)# 【解析】从回复中提取 Action 指令action self._parse_action(response)if error in action:print(f⚠️ {action[error]})self.messages.append({role: assistant, content: response})self.messages.append({role: user, content: fObservation: {action[error]}})continue# 【结束】LLM 认为任务完成if action[type] finish:print(f{*50}\n✅ 最终答案: {action[answer]}\n{*50})return action[answer]# 【行动】调用工具tool_info TOOL_REGISTRY.get(action[name])if not tool_info:obs f未知工具 {action[name]}可用: {list(TOOL_REGISTRY.keys())}else:try:result tool_info[func](**action[args])obs resultprint(f 调用 {action[name]}({action[args]}) → {result})except Exception as e:obs f工具执行错误: {e}# 【观察】将工具返回结果记入记忆供下一轮 LLM 使用self.messages.append({role: assistant, content: response})self.messages.append({role: user, content: fObservation: {obs}})print(f\n⚠️ 达到最大循环次数 {max_steps}自动停止)# ═══════════════════════════════════════════════════════════════# 3. 启动# 支持任何 OpenAI 兼容接口。# 常见配置取消注释即可使用# ═══════════════════════════════════════════════════════════════if __name__ __main__:# DeepSeek: modeldeepseek-chat, base_urlhttps://api.deepseek.com/v1agent Agent(modeldeepseek-v4-flash,api_keysk-your-api-key,base_urlhttps://api.deepseek.com/v1)agent.run(input( 请输入你的问题: ))