很多人想做 AI Agent,第一反应是上 LangChain、LangGraph 这类框架。框架能省事,但也会挡住你理解 Agent 到底是怎么跑起来的。本文用纯 Python 从零实现,不依赖任何 Agent 框架,把"感知—思考—行动"这个循环拆开讲清楚,最后接入真实大模型(DeepSeek / OpenAI 兼容 API)跑通工具调用。
读完你能得到两个可运行示例:
- 1. 一个最小原型,用规则大脑把 Agent 循环跑通;
- 2. 一个接入真实 LLM 的任务规划助手,实现"模型自己决定用哪个工具"的自动调度。
Agent 的核心:感知—思考—行动循环
把 AI Agent 看作一个"有大脑和手脚的程序",它由四部分构成:
- • 大脑(核心):负责理解、推理、决策的模型(LLM)。
- • 感知(输入):获取外部信息——用户指令、数据库结果、网页内容、传感器数据。
- • 行动(输出):影响外部世界——调用函数、发邮件、打印文本、控制设备。
- • 目标:所有行为围绕一个明确目标展开,例如"查一下北京明天的天气"。
感知-思考-行动循环这个循环就是 Agent 的骨架。下面两个示例都是在这套骨架上的不同实现。
示例一:最小原型,跑通循环
这个原型用一个"规则大脑"模拟 LLM,目的是先把循环结构写清楚。它包含五个组件:记忆、LLM 抽象、规划器、工具注册器、执行器。
先定义记忆(区分短期/长期)和 LLM 抽象(可替换点):
from typing import Any, Dict, List, Callableclass Memory: """非常轻量的记忆:short 存当前会话上下文,long 存长期偏好/联系人""" def __init__(self): self.short: Dict[str, Any] = {} self.long: Dict[str, Any] = {} def get_short(self, k, default=None): return self.short.get(k, default) def set_short(self, k, v): self.short[k] = v def get_long(self, k, default=None): return self.long.get(k, default) def set_long(self, k, v): self.long[k] = vclass LLMInterface: """LLM 抽象(替换点)。真实使用时改为 OpenAI/DeepSeek 调用。""" def generate(self, prompt: str) -> str: if "是否下雨" in prompt or "下雨" in prompt: return "请先查询天气;如果有雨,请生成提醒并发送给目标联系人。" if "生成提醒" in prompt: return "请提醒小王:明天北京有雨,请带伞。" return "我理解了。"
工具层用注册表统一管理,每个工具是独立函数:
class ToolRegistry: def __init__(self): self.tools: Dict[str, Callable] = {} def register(self, name: str, fn: Callable): self.tools[name] = fn def call(self, name: str, *args, **kwargs): if name not in self.tools: raise ValueError(f"工具未注册: {name}") return self.tools[name](*args, **kwargs)def mock_weather_api(city: str, date: str) -> Dict[str, Any]: if "北京" in city and "明天" in date: return {"city": city, "date": date, "cond": "雨", "precip_mm": 5} return {"city": city, "date": date, "cond": "晴", "precip_mm": 0}def mock_send_message(contact: str, message: str) -> bool: print(f"[发送消息] to={contact} message={message}") return Truedef mock_search(query: str) -> str: return f"模拟搜索结果:关于 `{query}` 的信息摘要。"
规划器把目标拆成有序步骤,执行器把步骤映射成工具调用:
class SimplePlanner: def plan(self, goal: str) -> List[Dict[str, Any]]: steps = [] if "天气" in goal or "下雨" in goal: steps.append({"action": "query_weather", "params": {"city": "北京", "date": "明天"}}) steps.append({"action": "decide_and_notify", "params": {"contact_name": "小王"}}) else: steps.append({"action": "search", "params": {"query": goal}}) return stepsclass Executor: def __init__(self, tools: ToolRegistry, memory: Memory, llm: LLMInterface): self.tools, self.memory, self.llm = tools, memory, llm def run_step(self, step: Dict[str, Any]): action = step["action"] params = step.get("params", {}) if action == "query_weather": res = self.tools.call("weather", params["city"], params["date"]) self.memory.set_short("last_weather", res) return res if action == "decide_and_notify": weather = self.memory.get_short("last_weather", {}) if weather.get("cond") == "雨": prompt = f"基于天气信息:{weather},生成一条发给{params['contact_name']}的提醒。" reminder = self.llm.generate(prompt) contact = self.memory.get_long(params["contact_name"]) or "13800000000" ok = self.tools.call("send_message", contact, reminder) return {"notified": ok, "message": reminder} return {"notified": False, "reason": "天气晴朗"} if action == "search": return self.tools.call("search", params["query"]) raise ValueError(f"未知动作: {action}")
Agent 本体把上述组件组装起来,并注册默认工具:
class SimpleAgent: def __init__(self): self.memory = Memory() self.tools = ToolRegistry() self.llm = LLMInterface() self.planner = SimplePlanner() self.executor = Executor(self.tools, self.memory, self.llm) self.tools.register("weather", mock_weather_api) self.tools.register("send_message", mock_send_message) self.tools.register("search", mock_search) self.memory.set_long("小王", "13911112222") def handle(self, user_prompt: str): intent = self.llm.generate(user_prompt) # 1) 大脑解析 steps = self.planner.plan(user_prompt) # 2) 规划 results = [] for step in steps: # 3) 逐步执行 results.append({"step": step, "result": self.executor.run_step(step)}) return {"intent": intent, "steps": results} # 4) 输出合并if __name__ == "__main__": agent = SimpleAgent() task = "查一下明天北京的天气,如果下雨,帮我写个提醒并发给小王。" import json print(json.dumps(agent.handle(task), ensure_ascii=False, indent=2))
这个原型跑通后,你就拥有了一个完整的"感知—思考—行动"循环。要接真实 LLM,只需把 LLMInterface.generate() 换成 API 调用;要加能力,只往 ToolRegistry 注册新函数。
⚠️ 注意:示例用规则模拟大脑,只为展示结构。生产环境的大脑必须是真实 LLM,工具执行要加失败重试与异常处理。
示例二:接入真实 LLM,实现自动工具调度
最小原型里"用哪个工具"是代码写死的。真实 Agent 应该让模型自己决定。下面用 DeepSeek(OpenAI 兼容 API)做大脑,构建一个任务规划助手。
先安装依赖:
pip install openai
申请 DeepSeek API Key:https://platform.deepseek.com/api_keys (OpenAI key 同样可用,把 base_url 换掉即可)。
大脑模块 brain.py:
from openai import OpenAIDEEPSEEK_API_KEY = "sk-xxxxxxx" # 替换为你的 keyDEEPSEEK_API_URL = "https://api.deepseek.com/v1"client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_API_URL)class AgentBrain: """Agent 的大脑,负责思考与决策""" def __init__(self, model="deepseek-v4-flash"): self.model = model def think(self, prompt): try: response = client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.5, # 越低越专注,适合规划 max_tokens=500, ) return response.choices[0].message.content.strip() except Exception as e: return f"思考过程出错: {e}"
temperature=0.5 在创造性和稳定性之间取平衡,规划类任务比聊天更依赖确定性。
工具模块 tools.py:
import datetimeclass AgentTools: @staticmethod def search_web(query): mock_results = { "旅行目的地推荐": "巴黎、东京、马尔代夫、云南丽江...", "北京天气": "明天晴,气温 15-25°C,微风。", "Python 教程": "推荐菜鸟教程等网站。", } for key, value in mock_results.items(): if key in query: return f"[网络搜索] 关于 '{query}' 的结果:{value}" return f"[网络搜索] 未找到 '{query}' 的明确信息。" @staticmethod def make_schedule(steps): schedule = "生成的日程计划:\n" for i, step in enumerate(steps, 1): schedule += f"{i}. {step}\n" return schedule @staticmethod def get_current_time(): now = datetime.datetime.now() return f"[系统时间] 现在是:{now.strftime('%Y-%m-%d %H:%M:%S')}" @staticmethod def calculate(expression): # ⚠️ 仅为演示 Agent 如何调用函数。eval 在生产中极危险,见下文安全提示。 try: allowed = set("0123456789+-*/(). ") if not all(c in allowed for c in expression): return "[计算器] 表达式包含非法字符,拒绝计算。" return f"[计算器] {expression} = {eval(expression, {'__builtins__': {}})}" except Exception as e: return f"[计算器] 计算错误: {str(e)}"
组装 Agent 的核心循环 agent.py。三个文件放在同一目录:
任务规划助手目录结构from brain import AgentBrainfrom tools import AgentToolsclass SimpleAgent: def __init__(self): self.brain = AgentBrain() self.tools = AgentTools self.tool_descriptions = """ 你可以使用以下工具: 1. 搜索工具:需要最新/未知信息时,例如'搜索 北京天气'。 2. 计划工具:整理多步骤计划时,例如'制定计划 [步骤1,步骤2]'。 3. 时间工具:需要知道当前时间,指令'获取时间'。 4. 计算工具:需要数学计算,例如'计算 3+5*2'。 """ def run(self, user_task): initial_prompt = f""" 你的角色是一个任务规划助手。 {self.tool_descriptions} 用户的任务是:{user_task} 请严格按照以下格式回答,不要添加额外内容: 思考:[简要分析任务需要什么] 工具:[选择工具名称,没有合适的写'无'] 指令:[发送给该工具的具体指令] """ initial_response = self.brain.think(initial_prompt) print("🧠 初始思考结果:") print(initial_response) # 解析思考结果,提取工具和指令 tool_to_use, tool_instruction = "无", "" for line in initial_response.split("\n"): line = line.strip() if line.startswith("工具:"): tool_to_use = line.replace("工具:", "").strip() elif line.startswith("指令:"): tool_instruction = line.replace("指令:", "").strip() result = self._use_tool(tool_to_use, tool_instruction) print("🔧 执行结果:") print(result) final_prompt = f""" 用户原始任务:{user_task} 思考过程:{initial_response} 工具执行结果:{result} 请生成一段对用户的最终回复,直接给出有帮助的答案或计划。 """ return self.brain.think(final_prompt) def _use_tool(self, tool_name, instruction): tool_name = tool_name.strip() if tool_name == "搜索工具": return self.tools.search_web(instruction) if tool_name == "计划工具": steps = [s.strip() for s in instruction.strip("[](){}").replace(",", ",").split(",") if s.strip()] return self.tools.make_schedule(steps) if tool_name == "时间工具": return self.tools.get_current_time() if tool_name == "计算工具": return self.tools.calculate(instruction) if tool_name == "无": return "[系统] 无需使用工具,直接回答用户即可。" return f"[系统] 未知工具:{tool_name},无法执行。"if __name__ == "__main__": my_agent = SimpleAgent() for task in [ "我想去旅行,帮我规划一下需要准备什么", "现在几点了?", "计算一下 15 的平方加上 20 的三分之一是多少", ]: print(my_agent.run(task)) print("\n" + "#" * 50 + "\n")
运行后会看到:模型先分析任务、选工具、给指令,代码执行工具,再把"思考+结果"喂回模型生成最终回复。这一步步就是 Agent 的自主循环。
权威补充:文本解析不是终点,用原生函数调用更稳
上面示例二靠解析"工具:/指令:"这种固定格式文本来调度,这是教学简化。Anthropic 在《Building Effective Agents》中把这类系统定义为 workflow(代码预定控制流)与 agent(模型自主决定控制流)的谱系,并建议:能用单次 LLM 调用加检索解决就不用 Agent,上框架前先用 LLM API 直接写。
生产级 Agent 用的是模型原生的 Function Calling(工具调用):你用 JSON Schema 描述函数,模型返回结构化的 tool_calls(函数名 + 参数),由你的代码执行。OpenAI 官方文档明确:
Function calling provides a way for models to interface with external systems... The model never executes code — it only returns a structured description of what to call and with what arguments. Your application code executes the actual function.
关键点:
- • 模型只决定调什么、传什么参,真正执行由你的代码控制,安全边界在你手里。
- • 用
strict: true(Structured Outputs)可强制返回严格符合 Schema 的 JSON,避免自由文本解析的不稳定。 - • 工具调用循环:用户提问 → 模型返回 tool_calls → 你的代码执行并回传结果 → 模型继续,直到给出最终答案。这正是示例二的逻辑,只是把"文本解析"换成了"结构化解析"。
进阶练习就是把示例二的文本格式改成让模型返回 JSON({"thought": "...", "tool": "...", "instruction": "..."}),用 json.loads() 解析,稳定性和可维护性会明显提升。
安全提示
tools.py 里的 calculate 用了 eval(),仅为演示 Agent 调用函数的机制。生产环境禁用 eval,它可执行任意代码(如 __import__('os').system('rm -rf /'))。改用 ast.literal_eval 或专门的数学表达式解析库,并始终对输入做白名单过滤。
总结
- • 核心认知:AI Agent = 大脑(LLM)+ 感知 + 行动 + 目标,本质是一个"感知—思考—行动"循环。
- • 最小原型:用记忆、LLM 抽象、工具注册器、规划器、执行器五个组件,规则大脑也能跑通循环,证明框架不是必需。
- • 真实接入:把 LLM 抽象换成 API 调用,让模型根据任务自主选择工具,循环由"文本解析"升级为"结构化工具调用"即可落地生产。
- • 延伸练习:① 加
conversation_history 实现记忆;② 用 JSON 输出替代文本解析;③ 给 AgentTools 加待办清单工具。
进一步学习:OpenAI Function Calling 文档(developers.openai.com)、Anthropic《Building Effective Agents》(anthropic.com/engineering/building-effective-agents)。