Python 实现 AI Agent

在深入代码之前，让我们先建立一个清晰的认知，可以把 AI Agent 想象成一个拥有大脑和手脚的智能程序。

  • 大脑（核心）：一个能够理解、推理和决策的模型（比如大型语言模型 LLM）。它负责处理信息，制定计划。
  • 感知（输入）：Agent 通过"感官"获取外部信息，如用户的指令、数据库查询结果、网页内容、传感器数据等。
  • 行动（输出）：Agent 通过"手脚"影响外部世界，例如调用一个函数、发送一封邮件、在屏幕上输出文本、控制机械臂等。
  • 目标：Agent 的一切行为都围绕着一个明确的目标展开，比如"帮我查一下北京明天的天气"。

其核心工作流程是一个经典的 "感知-思考-行动"循环，如下图所示：

简单的测试案例

agent-demo/
  agent.py        # 主实现（下面给出）
  tools.py        # 工具实现（可单独文件）
  memory.py       # 简单记忆实现（可单独文件）
  README.md

实例

"""
agent.py - 极简 AI Agent 原型（可直接运行）
说明：
- LLM 部分用一个非常简单的"规则回答器"模拟，实际接入 LLM 时替换 LLMInterface.generate(...)
- 包含组件：LLM 抽象、大脑(解析/决策)、规划器、工具注册器、内存、执行器
"""

from typing import Any, Dict, List, Callable
import time

# -----------------------------
# Memory（非常轻量）
# -----------------------------
class Memory:
    def __init__(self):
        self.short = {}   # 当前会话上下文
        self.long = {}    # 长期偏好/联系人等

    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] = v

# -----------------------------
# LLM 抽象（替换点）
# -----------------------------
class LLMInterface:
    def generate(self, prompt: str) -> str:
        """
        这里给出一个非常简单的规则式模拟回答器。
        真实使用时：替换为 OpenAI/其它模型的调用代码，返回 model 文本。
        """
        # 极简解析示例：识别是否需要判断"下雨"
        if "是否下雨" in prompt or "下雨" in prompt:
            return "请先查询天气；如果有雨，请生成提醒并发送给目标联系人。"
        if "生成提醒" in prompt:
            return "请提醒小王：明天北京有雨，请带伞。"
        return "我理解了。"

# -----------------------------
# 工具注册与模拟工具
# -----------------------------
class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, Callable[..., Any]] = {}

    def register(self, name: str, fn: Callable[..., Any]):
        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)

# 模拟工具：天气查询（真实情况会调用天气 API）
def mock_weather_api(city: str, date: str) -> Dict[str, Any]:
    # 简单规则：如果 city 包含 "北京" 且 date 包含 "明天"，返回下雨示例
    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 True

# 模拟工具：简单搜索（示意）
def mock_search(query: str) -> str:
    return f"模拟搜索结果：关于 `{query}` 的信息摘要。"

# -----------------------------
# Planner / Executor
# -----------------------------
class SimplePlanner:
    def plan(self, goal: str) -> List[Dict[str, Any]]:
        """
        将目标拆解为步骤列表（非常简化的实现）
        每一步包含：action(工具名或内部动作)、params
        """
        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 steps

class Executor:
    def __init__(self, tools: ToolRegistry, memory: Memory, llm: LLMInterface):
        self.tools = tools
        self.memory = memory
        self.llm = 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") == "雨":
                # 让 LLM 生成提醒文本（示例）
                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}
            else:
                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):
        # 1) 大脑解析（用 LLM 抽象）
        intent = self.llm.generate(user_prompt)
        # 2) 规划
        steps = self.planner.plan(user_prompt)
        # 3) 逐步执行
        results = []
        for step in steps:
            r = self.executor.run_step(step)
            results.append({"step": step, "result": r})
        # 4) 输出合并
        return {"intent": intent, "steps": results}

# -----------------------------
# 运行示例
# -----------------------------
if __name__ == "__main__":
    agent = SimpleAgent()
    task = "查一下明天北京的天气，如果下雨，帮我写个提醒并发给小王。"
    out = agent.handle(task)
    import json
    print(json.dumps(out, ensure_ascii=False, indent=2))

说明：

  1. LLM 抽象：LLMInterface.generate() 是可替换点。把这里换成对接 OpenAI、Claude 或本地 LLM 的代码即可。
  2. 规划器（Planner）：负责把自然语言目标拆成有序步骤。入门可以用规则或模板；进阶可用 LLM 输出步骤并解析为动作。
  3. 工具层（Tools）：每个工具是独立函数，统一注册到 ToolRegistry。真实项目中可能是 HTTP 请求、SDK 调用、数据库读写等。
  4. 记忆（Memory）：区分短期/长期。短期用于当前任务中间结果，长期用于联系人、偏好等。
  5. 执行器（Executor）：把步骤映射为工具调用并处理返回值、失败和重试逻辑。

构建你的第一个 AI Agent：任务规划助手

我们将构建一个 "任务规划助手"。它的目标是：根据用户给出的一个复杂任务（例如："我想去旅行"），自动将其分解成一系列有序的、可执行的具体步骤。

目录结构如下：

第一步：搭建智能"大脑"

Agent 的大脑需要推理能力。我们将使用 OpenAI 的 GPT 模型（通过 API 调用）作为我们 Agent 的"大脑"。

你需要准备一个 OpenAI API 密钥，也可以使用 DeepSeek 的 API key，可以兼容。

本章节测试示例使用了 DeepSeek 的 API key， 我们可以先去 https://platform.deepseek.com/api_keys 申请 。

首先，安装必要的库：

pip3 install openai

可以使用以下代码测试使用运行成功：

实例

import os
from openai import OpenAI


DEEPSEEK_API_KEY = "sk-xxxxxxx"  # 这里设置你申请的 key
DEEPSEEK_API_URL = "https://api.deepseek.com/v1"

client = OpenAI(
    api_key = DEEPSEEK_API_KEY,
    base_url = DEEPSEEK_API_URL)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Hello"},
    ],
    stream=False
)

print(response.choices[0].message.content)

如果安装配置成功会输出类似如下内容：

Hello! How can I assist you today?

然后，让我们创建一个最简单的"大脑"模块：

brain.py 文件代码

# brain.py
from openai import OpenAI
import os


DEEPSEEK_API_KEY = "sk-xxxxxxx"  # 这里设置你申请的 key
DEEPSEEK_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-chat"):
        self.model = model
   
    def think(self, prompt):
        """核心思考函数：接收提示，返回模型的思考结果"""
        try:
            # 使用客户端调用 Chat Completions API（v1.x 版本写法）
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.5,  # 控制创造性，越低越专注
                max_tokens=500    # 控制回复长度
            )
            # 提取模型返回的文本内容（v1.x 版本属性路径变更）
            reasoning = response.choices[0].message.content
            return reasoning.strip()
        except Exception as e:
            return f"思考过程出错: {e}"

# 简单测试一下大脑是否工作
if __name__ == "__main__":
    brain = AgentBrain()
    test_prompt = "你好，请简单介绍一下你自己。"
    print("测试提问：", test_prompt)
    print("大脑回复：", brain.think(test_prompt))

执行以上代码，输出结果大概如下：

测试提问： 你好，请简单介绍一下你自己。
大脑回复： 你好！我是DeepSeek，由深度求索公司创造的AI助手，很高兴认识你！😊

让我简单介绍一下自己：

**我的特点：**
**完全免费**：目前没有任何收费计划，你可以放心使用
**知识丰富**：知识截止到2024年7月，涵盖各个领域
**对话能力强**：支持128K上下文，可以进行长篇深入对话

代码解析：

  1. 我们创建了一个 AgentBrain 类来封装与 LLM 的交互。
  2. think 方法是核心，它接收一个字符串 prompt（提示词），将其发送给 GPT API。
  3. temperature 参数很重要：设置为 0.5 能让 Agent 在创造性和稳定性之间取得平衡，适合做规划任务。
  4. 最后返回模型生成的文本，这就是 Agent "思考"的结果。

第二步：定义"手脚"：工具与行动

Agent 需要有执行能力。我们为它定义几个简单的"工具"（函数），让它能够"动手"做事。

tools.py 文件代码

# tools.py
import datetime

class AgentTools:
    """Agent 可以使用的工具集合"""
   
    @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):
        """简单计算器工具（安全版本，修复兼容性问题）"""
        try:
            # 1. 严格过滤非法字符（仅允许数字、基础运算符、括号、空格）
            allowed_chars = set("0123456789+-*/(). ")
            if not all(c in allowed_chars for c in expression):
                return "[计算器] 表达式包含非法字符，拒绝计算。"
           
            # 2. 简化安全校验：移除易出错的 ast.walk 逻辑，改用基础语法检查
            # 先替换空格，避免空表达式
            clean_expr = expression.strip().replace(" ", "")
            if not clean_expr:
                return "[计算器] 表达式不能为空。"
           
            # 3. 安全执行计算（限制内置函数，仅保留基础运算）
            # 自定义安全的全局环境，仅允许基础数学运算
            safe_globals = {
                '__builtins__': {},
                'pow': pow,
                'abs': abs
            }
            # 使用 eval 但严格限制环境，且已过滤非法字符，大幅降低风险
            result = eval(clean_expr, safe_globals)
            return f"[计算器] {expression} = {result}"
       
        except ZeroDivisionError:
            return "[计算器] 错误：除数不能为零。"
        except SyntaxError:
            return "[计算器] 表达式语法错误（如括号不匹配、运算符错误）。"
        except Exception as e:
            return f"[计算器] 计算错误: {str(e)}"

# 测试工具
if __name__ == "__main__":
    print(AgentTools.search_web("北京天气"))
    print(AgentTools.get_current_time())
    print(AgentTools.calculate("3 + 5 * 2"))       # 正常计算
    print(AgentTools.calculate("10 / 0"))          # 除零错误
    print(AgentTools.calculate("__import__('os').system('ls')"))  # 非法字符过滤

代码解析：

  1. 每个 @staticmethod 都定义了一个 Agent 可以调用的"工具"。
  2. search_web 目前是模拟的，真实场景需要接入真正的搜索 API。
  3. make_schedule 用于将任务列表格式化为计划。
  4. 重要安全提示：calculate 工具中的 eval() 函数在真实产品中极其危险，必须用更安全的方式（如 ast.literal_eval）替换或彻底移除。这里仅为演示 Agent 如何调用函数。

第三步：组装 Agent：实现核心循环

现在，我们将"大脑"和"工具"组装起来，并实现核心的决策循环。我们将设计一个简单的规则：让大脑根据任务决定使用哪个工具。

实例

# agent.py
from brain import AgentBrain
from tools import AgentTools

class SimpleAgent:
    """一个简单的任务规划 AI Agent"""
   
    def __init__(self):
        self.brain = AgentBrain()
        # 静态工具类无需实例化
        self.tools = AgentTools
        # 定义 Agent 知道的工具列表及其描述，用于提示大脑
        self.tool_descriptions = """
        你可以使用以下工具：
        1. 搜索工具：当你需要获取最新、未知的信息时使用，例如'搜索 北京天气'。
        2. 计划工具：当你需要将多个步骤整理成计划时使用，例如'制定计划 [步骤1，步骤2]'。
        3. 时间工具：当你需要知道当前时间时使用，指令就是'获取时间'。
        4. 计算工具：当你需要进行数学计算时使用，例如'计算 3+5*2'。
        """
   
    def run(self, user_task):
        """运行 Agent 的主循环"""
        print(f"🎯 用户任务: {user_task}")
        print("=" * 40)
       
        # 第一步：感知与初步思考
        initial_prompt = f"""
        你的角色是一个任务规划助手。
        {self.tool_descriptions}
       
        用户的任务是：{user_task}
       
        请严格按照以下固定格式回答，不要添加额外内容：
        思考：[简要分析任务需要什么]
        工具：[选择要使用的工具名称，如果没有合适的就写'无']
        指令：[发送给该工具的具体指令内容]
        """
       
        initial_response = self.brain.think(initial_prompt)
        print("🧠 初始思考结果：")
        print(initial_response)
        print("-" * 20)
       
        # 第二步：解析思考结果，提取工具和指令（鲁棒性优化）
        lines = [line.strip() for line in initial_response.split('\n') if line.strip()]
        tool_to_use = "无"
        tool_instruction = ""
       
        for line in lines:
            if line.startswith("思考："):
                continue  # 跳过思考行
            elif 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)
        print("=" * 40)
       
        # 第四步：整合结果并反馈给用户
        final_prompt = f"""
        用户原始任务：{user_task}
        你已经进行了思考并使用了工具。
        思考过程：{initial_response}
        工具执行结果：{result}
       
        现在，请生成一段完整的、对用户的最终回复，直接给出有帮助的答案或计划，语言要自然友好。
        """
       
        final_response = self.brain.think(final_prompt)
        print("💡 最终回复给用户：")
        print(final_response)
       
        return final_response
   
    def _use_tool(self, tool_name, instruction):
        """根据工具名称调用具体的工具函数"""
        # 统一工具名称匹配（兼容大小写/多余空格）
        tool_name = tool_name.strip()
       
        if tool_name == "搜索工具":
            return self.tools.search_web(instruction)
        elif tool_name == "计划工具":
            # 兼容中英文逗号、括号，处理空指令
            if not instruction:
                return "[计划工具] 未提供计划步骤，无法生成日程。"
            # 移除括号并拆分步骤（兼容 []/()/{} 括号）
            clean_instr = instruction.strip('[](){}').strip()
            # 同时支持中英文逗号拆分
            steps = [s.strip() for s in clean_instr.replace('，', ',').split(',') if s.strip()]
            return self.tools.make_schedule(steps)
        elif tool_name == "时间工具":
            return self.tools.get_current_time()
        elif tool_name == "计算工具":
            return self.tools.calculate(instruction)
        elif tool_name == "无":
            return "[系统] 无需使用工具，直接回答用户即可。"
        else:
            return f"[系统] 未知工具：{tool_name}，无法执行。"

# 运行我们的 Agent！
if __name__ == "__main__":
    print("🤖 启动 SimpleAgent...")
    my_agent = SimpleAgent()
   
    # 测试几个不同的任务
    test_tasks = [
        "我想去旅行，帮我规划一下需要准备什么",
        "现在几点了？",
        "计算一下 15 的平方加上 20 的三分之一是多少",
    ]
   
    for task in test_tasks:
        my_agent.run(task)
        print("\n" + "#" * 50 + "\n")

执行代码，输出类似如下：

启动 SimpleAgent...
用户任务: 我想去旅行，帮我规划一下需要准备什么
========================================
初始思考结果：
思考：用户需要规划旅行前的准备工作。这是一个通用任务，不需要最新信息或计算，但需要将多个准备步骤整理成一个清晰的计划。

工具：计划工具

指令：制定计划 [1. 确定目的地和旅行时间，2. 预订交通票（机票/火车票）和住宿，3. 准备旅行证件（身份证、护照、签证等），4. 安排行程和景点门票，5. 打包行李（衣物、洗漱用品、药品、充电器等），6. 安排家中事务（宠物、植物、邮件等），7. 兑换货币或准备支付方式，8. 购买旅行保险]
--------------------
执行结果：
...

代码解析与运行流程：

  1. 初始化：Agent 创建了自己的大脑 (AgentBrain) 和工具箱 (AgentTools)。
  2. 接收任务：run 方法启动，接收 user_task（如"我想去旅行"）。
  3. 思考阶段：将任务和工具描述组合成 initial_prompt，发送给大脑。大脑返回的文本中包含了它选择的"工具"和给工具的"指令"。
  4. 行动阶段：_use_tool 函数根据大脑的选择，调用对应的工具函数并返回结果。
  5. 反馈与总结：将原始任务、思考过程和行动结果再次发送给大脑，让它生成一个对用户友好的最终回复。

运行这个程序，你会看到 Agent 如何一步步分析任务、选择工具、执行并给出答案。

实践练习：升级你的 Agent

你已经拥有了一个能跑通的 AI Agent！但这只是起点。尝试完成以下练习，让它变得更强大：

  1. 练习一：增加记忆功能

    • 目标：让 Agent 能记住对话历史。
    • 提示：在 SimpleAgent 类中添加一个 self.conversation_history = [] 列表。每次调用 brain.think 时，不仅发送当前提示，还将之前的历史对话也作为上下文发送。执行后，将新的对话回合（用户输入、AI 思考、工具结果）加入历史列表。

  2. 练习二：实现自动工具选择

    • 目标：让大脑的回复是标准的 JSON 格式，便于程序自动解析，而不是依赖文本提取。
    • 提示：修改提示词，要求大脑必须返回如 {"thought": "…"， "tool": "…"， "instruction": "…"} 的 JSON 字符串。然后在代码中使用 json.loads() 来解析，提高稳定性和准确性。

  3. 练习三：添加一个新工具

    • 目标：让 Agent 能处理待办事项。
    • 提示：在 AgentTools 类中添加 todo_list = [] 和两个新方法：add_todo(item) 和 show_todos()。在工具描述中告诉大脑这个新工具的存在，并更新 _use_tool 方法来调用它。

总结与展望

恭喜你！你已经成功构建了一个具备基本"感知-思考-行动"循环的 AI Agent。我们回顾一下关键点：

  • 核心概念：AI Agent = 大脑（LLM） + 感知 + 行动 + 目标。
  • 实现关键：
    1. 大脑：通过 API 调用 LLM（如 GPT）进行推理。
    2. 工具：将具体能力封装成函数，供 Agent 调用。
    3. 循环：用代码逻辑将思考、决策、执行串联起来。

这只是 AI Agent 世界的冰山一角。工业级框架如 LangChain、AutoGen 提供了更强大的工具集成、记忆管理、多 Agent 协作等功能。要深入探索，建议你：