摘要：本文将带你从零开始，用 Python 打造一个可运行的 AI Agent。我们会先解析 Agent 的核心原理，展示工作流程图，最后提供完整代码，让你能亲手体验 AI 自主完成任务的神奇过程。

---

一、AI Agent 是什么？

在 AI 大模型时代，“AI Agent”（人工智能代理）成为新的热点——它能像人类助手一样：

• 接收任务
• 规划步骤
• 调用工具
• 自主决策
• 甚至在遇到问题时自我修正

🤖 与普通聊天机器人的区别

普通聊天机器人	AI Agent
输入 → 模型 → 输出（线性）	目标 → 规划 → 执行 → 反馈 → 再规划（循环）
被动响应	主动思考和行动
无法调用外部工具	可集成多种工具函数

---

二、AI Agent 核心架构与流程图

2.1 标准 Agent 五模块架构

一个完整的 AI Agent 至少包含 5 个核心模块：

┌─────────────────┐
│    感知模块      │ ← 用户输入、环境信息
└────────┬────────┘
         ↓
┌─────────────────┐
│    记忆模块      │ ← 存储历史对话、上下文
└────────┬────────┘
         ↓
┌─────────────────┐
│    决策模块      │ ← LLM 核心，负责推理规划
└────────┬────────┘
         ↓
┌─────────────────┐
│    执行模块      │ ← 调用工具函数、执行动作
└────────┬────────┘
         ↓
┌─────────────────┐
│    反馈模块      │ ← 获取执行结果，形成闭环
└─────────────────┘

2.2 工作流程图

---

三、核心实现原理

AI Agent 的本质就是 “循环 + LLM + 工具函数”：

1. 循环机制：持续运行直到任务完成
2. LLM 核心：负责理解、推理、决策
3. 工具函数：扩展 Agent 的能力边界

关键在于 ReAct 框架（Reasoning + Acting）：

• Thought：思考当前状态和下一步行动
• Action：执行具体操作（调用工具）
• Observation：观察执行结果
• 重复直到任务完成

---

四、完整代码实现

4.1 环境准备

pip install openai python-dotenv requests

创建 .env 文件：

OPENAI_API_KEY=your_api_key_here

4.2 工具函数定义

# tools.py
import requests
import json
from datetime import datetime

class ToolRegistry:
    """工具注册器"""
    def __init__(self):
        self.tools = {}
    
    def register(self, name, func, description):
        """注册工具"""
        self.tools[name] = {
            'function': func,
            'description': description
        }
    
    def execute(self, tool_name, **kwargs):
        """执行工具"""
        if tool_name not in self.tools:
            return f"工具 '{tool_name}' 不存在"
        try:
            return self.tools[tool_name]['function'](**kwargs)
        except Exception as e:
            return f"执行工具时出错: {str(e)}"

# 具体工具实现
def get_current_time():
    """获取当前时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def search_weather(city):
    """模拟天气查询（实际项目中调用真实API）"""
    weather_data = {
        "北京": "晴天，25°C",
        "上海": "多云，28°C", 
        "广州": "雷阵雨，30°C",
        "深圳": "大雨，29°C"
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

def calculate_expression(expression):
    """安全的数学计算"""
    try:
        # 简单的安全检查，实际项目需要更严格的验证
        allowed_chars = set('0123456789+-*/(). ')
        if not all(c in allowed_chars for c in expression):
            return "表达式包含非法字符"
        result = eval(expression)
        return str(result)
    except:
        return "计算表达式无效"

# 注册工具
tool_registry = ToolRegistry()
tool_registry.register("get_current_time", get_current_time, "获取当前日期和时间")
tool_registry.register("search_weather", search_weather, "查询指定城市的天气，参数: city")
tool_registry.register("calculate_expression", calculate_expression, "计算数学表达式，参数: expression")

4.3 Agent 核心实现

# agent.py
import os
import json
from dotenv import load_dotenv
from openai import OpenAI
from tools import tool_registry

load_dotenv()

class SimpleAIAgent:
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = "gpt-4o-mini"  # 或使用 gpt-4-turbo
        self.conversation_history = []
        
    def get_tool_descriptions(self):
        """获取所有可用工具的描述"""
        descriptions = []
        for name, tool in tool_registry.tools.items():
            descriptions.append(f"- {name}: {tool['description']}")
        return "\n".join(descriptions)
    
    def create_system_prompt(self):
        """创建系统提示词"""
        return f"""你是一个智能AI助手，能够调用以下工具来完成任务：

可用工具：
{self.get_tool_descriptions()}

请按照以下格式进行响应：
1. 首先思考（Thought）你需要做什么
2. 如果需要调用工具，使用以下格式：
   Action: <工具名称>
   Action Input: <工具参数>
3. 如果不需要调用工具，直接给出最终答案（Final Answer）

记住：
- 仔细分析用户需求
- 选择最合适的工具
- 工具参数要准确
- 如果工具执行失败，尝试其他方法或告知用户"""
    
    def extract_action(self, response):
        """从LLM响应中提取动作"""
        if "Action:" in response and "Action Input:" in response:
            lines = response.split('\n')
            action = None
            action_input = None
            
            for line in lines:
                if line.startswith("Action:"):
                    action = line.replace("Action:", "").strip()
                elif line.startswith("Action Input:"):
                    action_input = line.replace("Action Input:", "").strip()
            
            if action and action_input:
                # 尝试解析 JSON 或直接使用字符串
                try:
                    input_dict = json.loads(action_input)
                    return action, input_dict
                except:
                    # 如果不是 JSON，当作单个字符串参数
                    return action, {"query": action_input}
        
        return None, None
    
    def run(self, user_input, max_steps=5):
        """运行Agent"""
        print(f"👤 用户: {user_input}\n")
        
        # 初始化对话历史
        messages = [
            {"role": "system", "content": self.create_system_prompt()},
            {"role": "user", "content": user_input}
        ]
        
        for step in range(max_steps):
            # 调用 LLM
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.7
            )
            
            llm_response = response.choices[0].message.content
            print(f"🧠 Agent 思考 (步骤 {step + 1}):")
            print(llm_response)
            print()
            
            # 检查是否是最终答案
            if "Final Answer:" in llm_response:
                final_answer = llm_response.split("Final Answer:")[-1].strip()
                print(f"✅ 最终答案: {final_answer}")
                return final_answer
            
            # 尝试提取动作
            action, action_input = self.extract_action(llm_response)
            
            if action and action in tool_registry.tools:
                # 执行工具
                print(f"🛠️  执行工具: {action}")
                print(f"📥 工具输入: {action_input}")
                
                tool_result = tool_registry.execute(action, **action_input)
                print(f"📤 工具输出: {tool_result}")
                print()
                
                # 将结果添加到对话历史
                messages.append({"role": "assistant", "content": llm_response})
                messages.append({"role": "user", "content": f"Observation: {tool_result}"})
                
            else:
                # 没有有效动作，直接返回
                print(f"✅ 直接回答: {llm_response}")
                return llm_response
        
        return "任务执行超时，请简化您的请求。"

# 使用示例
if __name__ == "__main__":
    agent = SimpleAIAgent()
    
    # 测试用例
    test_queries = [
        "现在几点了？",
        "北京今天天气怎么样？",
        "计算 123 * 456 + 789",
        "帮我查一下上海和广州的天气，然后告诉我哪个城市温度更高？"
    ]
    
    for query in test_queries:
        print("=" * 60)
        agent.run(query)
        print("\n" + "=" * 60 + "\n")

---

五、运行效果演示

示例 1：简单时间查询

👤 用户: 现在几点了？

🧠 Agent 思考 (步骤 1):
Thought: 我需要获取当前时间。
Action: get_current_time
Action Input: {}

🛠️  执行工具: get_current_time
📥 工具输入: {}
📤 工具输出: 2026-04-08 17:32:45

🧠 Agent 思考 (步骤 2):
Thought: 我已经获取了当前时间。
Final Answer: 现在是 2026-04-08 17:32:45。

✅ 最终答案: 现在是 2026-04-08 17:32:45。

示例 2：复杂多步任务

👤 用户: 帮我查一下上海和广州的天气，然后告诉我哪个城市温度更高？

🧠 Agent 思考 (步骤 1):
Thought: 我需要分别查询上海和广州的天气。
Action: search_weather  
Action Input: {"city": "上海"}

🛠️  执行工具: search_weather
📥 工具输入: {'city': '上海'}
📤 工具输出: 多云，28°C

🧠 Agent 思考 (步骤 2):
Thought: 我已经获取了上海的天气，现在需要查询广州的天气。
Action: search_weather
Action Input: {"city": "广州"}

🛠️  执行工具: search_weather
📥 工具输入: {'city': '广州'}
📤 工具输出: 雷阵雨，30°C

🧠 Agent 思考 (步骤 3):
Thought: 我已经获取了两个城市的天气数据，上海28°C，广州30°C。
Final Answer: 广州的温度更高，为30°C，而上海为28°C。

✅ 最终答案: 广州的温度更高，为30°C，而上海为28°C。

---

六、进阶优化方向

6.1 功能增强

• 长期记忆：集成向量数据库存储历史知识
• 多工具协调：处理更复杂的工具组合
• 错误恢复：自动重试或切换策略
• 用户确认：敏感操作前征求用户同意

6.2 性能优化

• 异步执行：支持并行工具调用
• 缓存机制：避免重复查询
• 流式响应：实时显示思考过程

6.3 安全考虑

• 工具沙箱：限制工具的权限范围
• 输入验证：防止恶意输入
• 审计日志：记录所有操作便于追踪

---

七、总结

通过本文的实践，我们实现了：

✅ 理解了 AI Agent 的核心原理
✅ 掌握了 ReAct 框架的工作机制
✅ 编写了完整的可运行代码
✅ 看到了实际的运行效果

AI Agent 的开发门槛其实并不高，核心就是 循环 + LLM + 工具函数 的组合。随着大模型能力的不断提升，Agent 将在更多场景中发挥重要作用。

动手建议：尝试添加更多工具函数，比如网页搜索、文件读写、邮件发送等，让你的 Agent 变得更强大！

---