当前位置:首页>python>Python 从零实现一个能思考、会调用工具的 AI Agent

Python 从零实现一个能思考、会调用工具的 AI Agent

  • 2026-08-25 06:29:50
Python 从零实现一个能思考、会调用工具的 AI Agent

很多人想做 AI Agent,第一反应是上 LangChain、LangGraph 这类框架。框架能省事,但也会挡住你理解 Agent 到底是怎么跑起来的。本文用纯 Python 从零实现,不依赖任何 Agent 框架,把"感知—思考—行动"这个循环拆开讲清楚,最后接入真实大模型(DeepSeek / OpenAI 兼容 API)跑通工具调用。

读完你能得到两个可运行示例:

  1. 1. 一个最小原型,用规则大脑把 Agent 循环跑通;
  2. 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)。

最新文章

随机文章