💻 完整代码 + 实战案例： GitHub 仓库
📖 配套教程： CSDN 专栏

💬 公众号：微信搜索「Lee 的成长日记」

如果觉得有用，欢迎 ⭐ Star 支持！

---

🎯 什么是 AI Agent？

大白话解释

场景：传统 AI vs Agent

传统 AI（ChatGPT）：
  你：帮我写个 Python 脚本
  AI：给你代码
  你：自己运行、调试、修改
  → AI 只是"顾问"

AI Agent：
  你：帮我分析这个 CSV 文件
  Agent：
    1. 读取文件
    2. 发现数据有问题
    3. 自动清洗
    4. 生成图表
    5. 保存报告
  → Agent 是"执行者"

核心区别：

LLM（大语言模型）：
  输入文本 → 输出文本
  → 被动响应

Agent（智能体）：
  接收任务 → 规划 → 调用工具 → 执行 → 验证 → 输出结果
  → 主动完成

正式定义

AI Agent = LLM + 记忆 + 工具 + 规划

┌─────────────────────────────┐
│         AI Agent            │
│                             │
│  ┌───────────────────────┐  │
│  │  LLM（大脑）           │  │  思考和决策
│  └───────────────────────┘  │
│           ↕                 │
│  ┌───────────────────────┐  │
│  │  Memory（记忆）        │  │  记住历史
│  └───────────────────────┘  │
│           ↕                 │
│  ┌───────────────────────┐  │
│  │  Tools（工具）         │  │  调用外部能力
│  └───────────────────────┘  │
│           ↕                 │
│  ┌───────────────────────┐  │
│  │  Planning（规划）      │  │  拆解任务
│  └───────────────────────┘  │
└─────────────────────────────┘

---

1️⃣ Agent 能做什么？

能力 1：自动化工作流

场景：每日数据报告

"""
传统方式：
  每天早上手动：
  1. 登录数据库
  2. 导出 CSV
  3. 用 Excel 画图
  4. 写邮件发送
  → 耗时 30 分钟

Agent 方式：
  设置定时任务
  Agent 自动完成所有步骤
  → 耗时 0 分钟
"""

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
import pandas as pd
import smtplib
from email.mime.text import MIMEText

class ReportAgent:
    """日报生成 Agent"""
    
    def __init__(self):
        self.tools = [
            Tool(
                name="query_database",
                func=self.query_database,
                description="查询数据库获取销售数据"
            ),
            Tool(
                name="generate_chart",
                func=self.generate_chart,
                description="根据数据生成图表"
            ),
            Tool(
                name="send_email",
                func=self.send_email,
                description="发送邮件报告"
            )
        ]
    
    def query_database(self, query: str) -> str:
        """查询数据库"""
        # 实际应该连接真实数据库
        df = pd.read_csv("sales_data.csv")
        return df.to_string()
    
    def generate_chart(self, data: str) -> str:
        """生成图表"""
        import matplotlib.pyplot as plt
        
        df = pd.read_csv("sales_data.csv")
        
        plt.figure(figsize=(10, 6))
        plt.plot(df['date'], df['sales'])
        plt.title('Daily Sales')
        plt.savefig('sales_chart.png')
        
        return "Chart saved to sales_chart.png"
    
    def send_email(self, content: str) -> str:
        """发送邮件"""
        msg = MIMEText(content)
        msg['Subject'] = 'Daily Sales Report'
        msg['From'] = 'agent@company.com'
        msg['To'] = 'manager@company.com'
        
        # 实际应该配置 SMTP
        # with smtplib.SMTP('smtp.company.com') as server:
        #     server.send_message(msg)
        
        return "Email sent successfully"
    
    def run(self):
        """执行日报生成"""
        print("🤖 Agent 开始生成日报...")
        
        # 步骤 1：查询数据
        print("📊 查询数据库...")
        data = self.query_database("SELECT * FROM sales WHERE date = TODAY")
        
        # 步骤 2：生成图表
        print("📈 生成图表...")
        chart_result = self.generate_chart(data)
        
        # 步骤 3：发送邮件
        print("📧 发送邮件...")
        email_result = self.send_email(f"Report generated.\n{chart_result}")
        
        print("✅ 日报生成完成！")


# 使用
agent = ReportAgent()
agent.run()

---

能力 2：多步推理与纠错

场景：复杂问题求解

用户：计算 (123 + 456) * 789 / 12 的结果，并判断是否为质数

传统 LLM：
  可能算错（LLM 不擅长数学）

Agent：
  步骤 1：调用计算器工具计算 (123 + 456) * 789 / 12 = 37972.25
  步骤 2：判断 37972.25 不是整数，所以不是质数
  步骤 3：返回结果
  → 准确
"""

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI

def calculator(expression: str) -> str:
    """安全计算器"""
    try:
        # 只允许数学运算
        allowed_chars = set('0123456789+-*/(). ')
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters"
        
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

def is_prime(n: str) -> str:
    """判断是否为质数"""
    try:
        num = float(n)
        
        # 不是整数
        if num != int(num):
            return f"{num} is not an integer, so not prime"
        
        num = int(num)
        
        if num < 2:
            return f"{num} is not prime"
        
        for i in range(2, int(num**0.5) + 1):
            if num % i == 0:
                return f"{num} is not prime (divisible by {i})"
        
        return f"{num} is prime"
    
    except Exception as e:
        return f"Error: {str(e)}"

# 创建工具
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="Useful for mathematical calculations"
    ),
    Tool(
        name="PrimeChecker",
        func=is_prime,
        description="Check if a number is prime"
    )
]

# 初始化 Agent
llm = OpenAI(temperature=0)
agent = initialize_agent(
    tools, 
    llm, 
    agent="zero-shot-react-description",
    verbose=True
)

# 执行
result = agent.run("Calculate (123 + 456) * 789 / 12 and check if it's prime")
print(result)

输出：

> Entering new AgentExecutor chain...
I need to calculate the expression first, then check if the result is prime.
Action: Calculator
Action Input: (123 + 456) * 789 / 12
Observation: 37972.25
Thought: The result is 37972.25, which is not an integer.
Action: PrimeChecker
Action Input: 37972.25
Observation: 37972.25 is not an integer, so not prime
Thought: I now know the final answer.
Final Answer: The result is 37972.25, which is not an integer, so it's not a prime number.
> Finished chain.

---

能力 3：自主学习和适应

场景：客服机器人

class LearningCustomerServiceAgent:
    """学习型客服 Agent"""
    
    def __init__(self):
        self.knowledge_base = {}  # 知识库
        self.conversation_history = []  # 对话历史
        self.success_rate = {}  # 问题解决率
    
    def handle_query(self, query: str) -> str:
        """处理用户咨询"""
        # 步骤 1：检索知识库
        answer = self.search_knowledge_base(query)
        
        if answer:
            # 找到答案
            self.record_success(query)
            return answer
        
        else:
            # 步骤 2：调用 LLM 生成答案
            answer = self.generate_answer_with_llm(query)
            
            # 步骤 3：人工审核（首次）
            if self.needs_human_review(query):
                print(f"⚠️  需要人工审核：{query}")
                approved = input("是否批准此答案？(y/n): ")
                
                if approved.lower() == 'y':
                    # 加入知识库
                    self.add_to_knowledge_base(query, answer)
                    self.record_success(query)
                else:
                    self.record_failure(query)
                    return "抱歉，我需要进一步学习才能回答这个问题。"
            else:
                # 直接返回
                self.add_to_knowledge_base(query, answer)
                self.record_success(query)
            
            return answer
    
    def search_knowledge_base(self, query: str) -> str:
        """检索知识库"""
        # 简化版：关键词匹配
        for key, value in self.knowledge_base.items():
            if key.lower() in query.lower():
                return value
        return None
    
    def add_to_knowledge_base(self, query: str, answer: str):
        """添加到知识库"""
        self.knowledge_base[query] = answer
        print(f"✅ 已学习新问题：{query[:50]}...")
    
    def generate_answer_with_llm(self, query: str) -> str:
        """用 LLM 生成答案"""
        from openai import OpenAI
        
        client = OpenAI(api_key="your-api-key")
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个专业的客服助手"},
                {"role": "user", "content": query}
            ]
        )
        
        return response.choices[0].message.content
    
    def needs_human_review(self, query: str) -> bool:
        """判断是否需要人工审核"""
        # 如果是新问题，需要审核
        return query not in self.knowledge_base
    
    def record_success(self, query: str):
        """记录成功解决"""
        self.success_rate[query] = self.success_rate.get(query, 0) + 1
    
    def record_failure(self, query: str):
        """记录失败"""
        if query in self.success_rate:
            del self.success_rate[query]
    
    def get_statistics(self) -> dict:
        """获取统计信息"""
        total_queries = len(self.conversation_history)
        successful_queries = sum(self.success_rate.values())
        
        return {
            "total_queries": total_queries,
            "successful_queries": successful_queries,
            "success_rate": successful_queries / total_queries if total_queries > 0 else 0,
            "knowledge_size": len(self.knowledge_base)
        }


# 使用
agent = LearningCustomerServiceAgent()

# 模拟对话
queries = [
    "怎么重置密码？",
    "如何退款？",
    "怎么重置密码？",  # 第二次问，直接从知识库回答
    "产品保修期多久？"
]

for query in queries:
    print(f"\n用户：{query}")
    answer = agent.handle_query(query)
    print(f"Agent：{answer[:100]}...")

# 查看统计
stats = agent.get_statistics()
print(f"\n📊 统计信息：")
print(f"  总查询数：{stats['total_queries']}")
print(f"  成功解决：{stats['successful_queries']}")
print(f"  成功率：{stats['success_rate']:.2%}")
print(f"  知识库大小：{stats['knowledge_size']}")

---

2️⃣ Agent 不能做什么？

限制 1：无法真正"理解"

Agent：
  ✅ 可以执行任务
  ✅ 可以调用工具
  ❌ 但不理解自己在做什么

示例：
  Agent 可以写代码、运行测试、修复 Bug
  但它不知道"为什么这段代码更好"
  → 只是模式匹配

实际影响：

# Agent 可能生成这样的代码：
def calculate(x, y):
    return x + y  # 正确，但不知道为什么

# 人类程序员会这样：
def calculate_total_price(price: float, quantity: int) -> float:
    """
    计算总价
    
    Args:
        price: 单价
        quantity: 数量
    
    Returns:
        总价
    """
    return price * quantity  # 有意义的命名和文档

---

限制 2：依赖工具质量

Agent 的能力上限 = 可用工具的能力

如果工具不好：
  - 数据库查询慢 → Agent 慢
  - API 不稳定 → Agent 失败
  - 工具功能有限 → Agent 能力受限

示例：

# 糟糕的工具设计
def search_web(query: str) -> str:
    """搜索网络（返回整个 HTML）"""
    import requests
    response = requests.get(f"https://www.google.com/search?q={query}")
    return response.text  # 返回几 MB 的 HTML

# 问题：
# 1. Token 爆炸（LLM 上下文不够）
# 2. 信息噪音大（HTML 标签干扰）
# 3. 速度慢

# 好的工具设计
def search_web(query: str) -> str:
    """搜索网络（返回摘要）"""
    from serpapi import GoogleSearch
    
    params = {
        "q": query,
        "num": 3,  # 只要前 3 条结果
        "api_key": "your-api-key"
    }
    
    search = GoogleSearch(params)
    results = search.get_dict()
    
    # 提取标题和摘要
    snippets = []
    for result in results.get("organic_results", [])[:3]:
        snippets.append(f"Title: {result['title']}\nSnippet: {result['snippet']}")
    
    return "\n\n".join(snippets)  # 简洁、结构化

---

限制 3：长链条容易出错

Agent 执行步骤越多，出错概率越高

单步任务：
  成功率：95%

5 步任务：
  成功率：0.95^5 = 77%

10 步任务：
  成功率：0.95^10 = 60%

20 步任务：
  成功率：0.95^20 = 36%  ← 太低了！

解决方案：

class RobustAgent:
    """鲁棒性 Agent"""
    
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
    
    def execute_with_retry(self, task: callable, *args, **kwargs):
        """带重试的执行"""
        for attempt in range(self.max_retries):
            try:
                result = task(*args, **kwargs)
                
                # 验证结果
                if self.validate_result(result):
                    return result
                else:
                    print(f"⚠️  结果验证失败，重试 {attempt + 1}/{self.max_retries}")
            
            except Exception as e:
                print(f"❌ 执行失败：{e}，重试 {attempt + 1}/{self.max_retries}")
        
        raise Exception(f"任务失败，已重试 {self.max_retries} 次")
    
    def validate_result(self, result) -> bool:
        """验证结果合理性"""
        # 根据任务类型定制验证逻辑
        if isinstance(result, str):
            return len(result) > 0  # 非空
        
        elif isinstance(result, dict):
            return "error" not in result  # 无错误
        
        return True  # 默认通过
    
    def break_down_task(self, complex_task: str) -> List[str]:
        """拆解复杂任务为小任务"""
        # 调用 LLM 拆解
        from openai import OpenAI
        
        client = OpenAI(api_key="your-api-key")
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "将复杂任务拆解为不超过 3 步的子任务"},
                {"role": "user", "content": complex_task}
            ]
        )
        
        # 解析返回的子任务列表
        subtasks = response.choices[0].message.content.split("\n")
        return [t.strip() for t in subtasks if t.strip()]

---

限制 4：安全性风险

Agent 可以执行命令 → 可能被滥用

风险：
  1. 注入攻击（Prompt Injection）
  2. 无限循环（资源耗尽）
  3. 敏感数据泄露
  4. 误操作（删除重要文件）

防护措施：

class SecureAgent:
    """安全 Agent"""
    
    def __init__(self):
        self.allowed_commands = {
            "ls", "cat", "grep", "python", "pytest"
        }
        self.forbidden_patterns = [
            "rm -rf", "sudo", "chmod 777", "DROP TABLE"
        ]
        self.execution_timeout = 60  # 60 秒超时
        self.max_iterations = 10  # 最多 10 次迭代
    
    def validate_command(self, command: str) -> bool:
        """验证命令安全性"""
        # 检查是否在白名单
        base_command = command.split()[0]
        if base_command not in self.allowed_commands:
            print(f"❌ 不允许的命令：{base_command}")
            return False
        
        # 检查危险模式
        for pattern in self.forbidden_patterns:
            if pattern in command:
                print(f"❌ 检测到危险模式：{pattern}")
                return False
        
        return True
    
    def execute_safely(self, command: str) -> str:
        """安全执行命令"""
        # 验证
        if not self.validate_command(command):
            raise SecurityError("Command blocked")
        
        # 执行（带超时）
        import subprocess
        
        try:
            result = subprocess.run(
                command.split(),
                capture_output=True,
                text=True,
                timeout=self.execution_timeout
            )
            
            return result.stdout
        
        except subprocess.TimeoutExpired:
            raise SecurityError("Command timed out")
    
    def detect_prompt_injection(self, user_input: str) -> bool:
        """检测 Prompt 注入攻击"""
        injection_patterns = [
            "ignore previous instructions",
            "system prompt",
            "you are now",
            "override security"
        ]
        
        user_input_lower = user_input.lower()
        
        for pattern in injection_patterns:
            if pattern in user_input_lower:
                print(f"⚠️  检测到潜在的 Prompt 注入：{pattern}")
                return True
        
        return False
    
    def process_request(self, user_input: str) -> str:
        """处理用户请求（带安全检查）"""
        # 检查注入攻击
        if self.detect_prompt_injection(user_input):
            return "抱歉，我无法处理这个请求。"
        
        # 正常处理
        return self.execute_task(user_input)


class SecurityError(Exception):
    """安全错误"""
    pass

---

3️⃣ Agent 的核心技术

技术 1：ReAct 框架（Reasoning + Acting）

原理：

传统 LLM：
  输入 → 思考 → 输出

ReAct：
  输入 → 思考 → 行动 → 观察 → 思考 → 行动 → ... → 输出

代码实现：

class ReActAgent:
    """ReAct Agent 实现"""
    
    def __init__(self, llm, tools: List[Tool]):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = 10
    
    def run(self, task: str) -> str:
        """执行任务"""
        history = []
        
        for i in range(self.max_iterations):
            # 步骤 1：思考（Thought）
            thought = self.think(task, history)
            history.append(f"Thought: {thought}")
            
            # 步骤 2：决定行动（Action）
            action_name, action_input = self.decide_action(thought)
            
            if action_name == "FINISH":
                # 任务完成
                return action_input
            
            history.append(f"Action: {action_name}({action_input})")
            
            # 步骤 3：执行行动
            observation = self.execute_action(action_name, action_input)
            history.append(f"Observation: {observation}")
        
        raise Exception("Max iterations reached")
    
    def think(self, task: str, history: List[str]) -> str:
        """思考下一步"""
        prompt = f"""
Task: {task}

History:
{chr(10).join(history)}

What should I do next? Think step by step.
"""
        
        response = self.llm.generate(prompt)
        return response
    
    def decide_action(self, thought: str) -> Tuple[str, str]:
        """决定行动"""
        # 调用 LLM 决定
        prompt = f"""
Based on this thought: {thought}

Choose an action from: {list(self.tools.keys())}
Or output FINISH if task is complete.

Format: ACTION_NAME(ACTION_INPUT)
"""
        
        response = self.llm.generate(prompt)
        
        # 解析
        if "FINISH" in response:
            return "FINISH", response.replace("FINISH:", "").strip()
        
        # 解析 ACTION(INPUT)
        import re
        match = re.match(r'(\w+)\((.+)\)', response)
        
        if match:
            return match.group(1), match.group(2)
        
        raise Exception(f"Invalid action format: {response}")
    
    def execute_action(self, action_name: str, action_input: str) -> str:
        """执行行动"""
        if action_name not in self.tools:
            return f"Error: Unknown action {action_name}"
        
        tool = self.tools[action_name]
        
        try:
            result = tool.func(action_input)
            return str(result)
        except Exception as e:
            return f"Error: {str(e)}"

---

技术 2：Memory（记忆系统）

三层记忆架构：

短期记忆（Short-term Memory）：
  - 当前对话历史
  - 容量：最近 10-20 轮
  - 用途：保持上下文连贯

长期记忆（Long-term Memory）：
  - 向量数据库存储
  - 容量：无限
  - 用途：检索历史信息

工作记忆（Working Memory）：
  - 当前任务的中间状态
  - 容量：有限
  - 用途：任务执行中的临时变量

代码实现：

import chromadb
from chromadb.config import Settings

class MemorySystem:
    """记忆系统"""
    
    def __init__(self):
        # 短期记忆
        self.short_term_memory = []
        self.max_short_term = 20
        
        # 长期记忆（向量数据库）
        self.client = chromadb.Client(Settings())
        self.long_term_collection = self.client.create_collection(
            name="long_term_memory",
            metadata={"description": "Agent's long-term memory"}
        )
        
        # 工作记忆
        self.working_memory = {}
    
    def add_short_term(self, message: str, role: str = "user"):
        """添加短期记忆"""
        self.short_term_memory.append({
            "role": role,
            "content": message,
            "timestamp": time.time()
        })
        
        # 限制长度
        if len(self.short_term_memory) > self.max_short_term:
            # 压缩到长期记忆
            old_messages = self.short_term_memory[:-10]
            self.compress_to_long_term(old_messages)
            
            # 保留最近 10 条
            self.short_term_memory = self.short_term_memory[-10:]
    
    def compress_to_long_term(self, messages: List[Dict]):
        """压缩到长期记忆"""
        # 总结消息
        summary = self.summarize_messages(messages)
        
        # 存入向量数据库
        self.long_term_collection.add(
            documents=[summary],
            metadatas=[{"type": "conversation_summary"}],
            ids=[f"conv_{int(time.time())}"]
        )
    
    def retrieve_from_long_term(self, query: str, n_results: int = 3) -> List[str]:
        """从长期记忆检索"""
        results = self.long_term_collection.query(
            query_texts=[query],
            n_results=n_results
        )
        
        return results['documents'][0]
    
    def set_working_memory(self, key: str, value: any):
        """设置工作记忆"""
        self.working_memory[key] = value
    
    def get_working_memory(self, key: str) -> any:
        """获取工作记忆"""
        return self.working_memory.get(key)
    
    def clear_working_memory(self):
        """清空工作记忆"""
        self.working_memory.clear()
    
    def summarize_messages(self, messages: List[Dict]) -> str:
        """总结消息（调用 LLM）"""
        from openai import OpenAI
        
        client = OpenAI(api_key="your-api-key")
        
        messages_text = "\n".join([
            f"{m['role']}: {m['content']}" for m in messages
        ])
        
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "总结以下对话，提取关键信息"},
                {"role": "user", "content": messages_text}
            ]
        )
        
        return response.choices[0].message.content
    
    def get_context(self, current_query: str) -> str:
        """获取完整上下文"""
        # 短期记忆
        short_term_context = "\n".join([
            f"{m['role']}: {m['content']}" 
            for m in self.short_term_memory[-10:]
        ])
        
        # 长期记忆（相关检索）
        long_term_context = self.retrieve_from_long_term(current_query)
        
        # 工作记忆
        working_memory_str = "\n".join([
            f"{k}: {v}" for k, v in self.working_memory.items()
        ])
        
        return f"""
Short-term Memory:
{short_term_context}

Long-term Memory (relevant):
{chr(10).join(long_term_context)}

Working Memory:
{working_memory_str}
"""

---

技术 3：Tool Use（工具调用）

Function Calling 原理：

传统方式：
  用户：今天北京天气如何？
  LLM：我不知道（没有实时数据）

Function Calling：
  用户：今天北京天气如何？
  LLM：我需要调用 weather_api
  → 系统执行 weather_api("北京")
  → 返回：{"temp": 25, "condition": "sunny"}
  LLM：今天北京天气晴朗，气温 25°C

代码实现：

import json
from typing import Dict, List, Callable

class ToolRegistry:
    """工具注册表"""
    
    def __init__(self):
        self.tools = {}
    
    def register(self, name: str, func: Callable, description: str, 
                 parameters: Dict):
        """注册工具"""
        self.tools[name] = {
            "function": func,
            "description": description,
            "parameters": parameters
        }
    
    def get_tool_schema(self) -> List[Dict]:
        """获取工具 schema（用于 LLM）"""
        schemas = []
        
        for name, tool in self.tools.items():
            schemas.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": tool["description"],
                    "parameters": tool["parameters"]
                }
            })
        
        return schemas
    
    def execute(self, tool_name: str, arguments: Dict) -> any:
        """执行工具"""
        if tool_name not in self.tools:
            raise ValueError(f"Unknown tool: {tool_name}")
        
        tool = self.tools[tool_name]
        
        try:
            result = tool["function"](**arguments)
            return result
        except Exception as e:
            return f"Error: {str(e)}"


# 注册工具
registry = ToolRegistry()

# 天气查询
def get_weather(city: str) -> Dict:
    """获取天气信息"""
    # 实际应该调用天气 API
    return {
        "city": city,
        "temperature": 25,
        "condition": "sunny",
        "humidity": 60
    }

registry.register(
    name="get_weather",
    func=get_weather,
    description="Get current weather for a city",
    parameters={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "City name"
            }
        },
        "required": ["city"]
    }
)

# 计算器
def calculate(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

registry.register(
    name="calculate",
    func=calculate,
    description="Calculate a mathematical expression",
    parameters={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "Mathematical expression"
            }
        },
        "required": ["expression"]
    }
)

# 使用
from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# 用户提问
user_query = "今天北京天气如何？如果温度超过 30 度，计算 (30 - 25) * 2"

# 第一步：LLM 决定调用哪些工具
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": user_query}],
    tools=registry.get_tool_schema(),
    tool_choice="auto"
)

# 检查是否有工具调用
if response.choices[0].message.tool_calls:
    for tool_call in response.choices[0].message.tool_calls:
        tool_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        print(f"🔧 调用工具：{tool_name}")
        print(f"   参数：{arguments}")
        
        # 执行工具
        result = registry.execute(tool_name, arguments)
        print(f"   结果：{result}")
        
        # 第二步：LLM 根据工具结果生成最终回答
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "user", "content": user_query},
                {"role": "assistant", "content": None, "tool_calls": [tool_call]},
                {"role": "tool", "tool_call_id": tool_call.id, "content": str(result)}
            ]
        )
        
        print(f"\n💬 最终回答：{final_response.choices[0].message.content}")

---

技术 4：Planning（任务规划）

Tree of Thoughts（思维树）：

传统线性思考：
  问题 → 思路 1 → 思路 2 → 答案

思维树：
  问题
  ├─ 思路 1
  │   ├─ 子思路 1.1 → 评估：好
  │   └─ 子思路 1.2 → 评估：差
  ├─ 思路 2
  │   ├─ 子思路 2.1 → 评估：一般
  │   └─ 子思路 2.2 → 评估：好 ✓
  └─ 思路 3
      └─ 子思路 3.1 → 评估：差

选择最佳路径：思路 2 → 子思路 2.2

代码实现：

class TreeOfThoughtsAgent:
    """思维树 Agent"""
    
    def __init__(self, llm, branching_factor: int = 3, depth: int = 3):
        self.llm = llm
        self.branching_factor = branching_factor  # 每个节点分支数
        self.depth = depth  # 树的深度
    
    def solve(self, problem: str) -> str:
        """解决问题"""
        # 构建思维树
        tree = self.build_tree(problem)
        
        # 评估所有路径
        best_path = self.evaluate_paths(tree)
        
        # 返回最佳路径的答案
        return best_path['answer']
    
    def build_tree(self, problem: str) -> Dict:
        """构建思维树"""
        root = {
            "thought": problem,
            "children": [],
            "value": 0
        }
        
        # BFS 构建
        queue = [(root, 0)]  # (node, depth)
        
        while queue:
            node, current_depth = queue.pop(0)
            
            if current_depth >= self.depth:
                continue
            
            # 生成子节点
            children_thoughts = self.generate_thoughts(node['thought'])
            
            for thought in children_thoughts:
                child = {
                    "thought": thought,
                    "children": [],
                    "value": 0
                }
                
                node['children'].append(child)
                queue.append((child, current_depth + 1))
        
        return root
    
    def generate_thoughts(self, current_thought: str) -> List[str]:
        """生成多个可能的思路"""
        prompt = f"""
Current thought: {current_thought}

Generate {self.branching_factor} different approaches to continue.
Each approach should be concise (1-2 sentences).

Output format:
1. Approach 1
2. Approach 2
3. Approach 3
"""
        
        response = self.llm.generate(prompt)
        
        # 解析
        thoughts = []
        for line in response.split('\n'):
            if line.strip() and line[0].isdigit():
                thought = line.split('.', 1)[1].strip()
                thoughts.append(thought)
        
        return thoughts[:self.branching_factor]
    
    def evaluate_paths(self, tree: Dict) -> Dict:
        """评估所有路径，找到最佳"""
        all_paths = []
        self.collect_paths(tree, [], all_paths)
        
        # 评估每条路径
        for path in all_paths:
            path['value'] = self.evaluate_path(path['thoughts'])
        
        # 选择最佳
        best_path = max(all_paths, key=lambda x: x['value'])
        
        return best_path
    
    def collect_paths(self, node: Dict, current_path: List, all_paths: List):
        """收集所有路径"""
        current_path.append(node['thought'])
        
        if not node['children']:
            # 叶子节点，保存路径
            all_paths.append({
                'thoughts': current_path.copy(),
                'answer': current_path[-1],
                'value': 0
            })
        else:
            for child in node['children']:
                self.collect_paths(child, current_path, all_paths)
        
        current_path.pop()
    
    def evaluate_path(self, thoughts: List[str]) -> float:
        """评估路径质量"""
        # 调用 LLM 评分
        prompt = f"""
Evaluate the quality of this reasoning path (0-10):

{' → '.join(thoughts)}

Score:
"""
        
        response = self.llm.generate(prompt)
        
        # 提取分数
        import re
        match = re.search(r'(\d+\.?\d*)', response)
        
        if match:
            return float(match.group(1))
        
        return 0.0

---

4️⃣ 前沿技术

技术 1：Multi-Agent（多智能体协作）

场景：软件开发团队

传统单 Agent：
  一个 Agent 做所有事
  → 容易出错，效率低

Multi-Agent：
  - Product Manager Agent（产品经理）
  - Developer Agent（开发者）
  - Tester Agent（测试工程师）
  - Reviewer Agent（代码审查）
  
  协作完成项目
  → 专业化，效率高

代码实现：

class MultiAgentSystem:
    """多智能体系统"""
    
    def __init__(self):
        self.agents = {
            "product_manager": ProductManagerAgent(),
            "developer": DeveloperAgent(),
            "tester": TesterAgent(),
            "reviewer": ReviewerAgent()
        }
    
    def execute_project(self, requirement: str) -> Dict:
        """执行项目"""
        print("🚀 启动项目...\n")
        
        # 步骤 1：产品经理分析需求
        print("📋 产品经理分析需求...")
        spec = self.agents["product_manager"].analyze(requirement)
        print(f"   需求规格：{spec[:100]}...\n")
        
        # 步骤 2：开发者实现
        print("💻 开发者编写代码...")
        code = self.agents["developer"].implement(spec)
        print(f"   代码行数：{len(code.split(chr(10)))} 行\n")
        
        # 步骤 3：测试工程师测试
        print("🧪 测试工程师测试...")
        test_result = self.agents["tester"].test(code)
        print(f"   测试结果：{'通过' if test_result['passed'] else '失败'}\n")
        
        if not test_result['passed']:
            # 返回修复
            print("🔧 返回修复...\n")
            code = self.agents["developer"].fix(code, test_result['errors'])
        
        # 步骤 4：代码审查
        print("👀 代码审查...")
        review = self.agents["reviewer"].review(code)
        print(f"   审查意见：{review[:100]}...\n")
        
        return {
            "code": code,
            "test_result": test_result,
            "review": review
        }


class ProductManagerAgent:
    """产品经理 Agent"""
    def analyze(self, requirement: str) -> str:
        # 调用 LLM 生成需求规格
        return "Detailed specification..."


class DeveloperAgent:
    """开发者 Agent"""
    def implement(self, spec: str) -> str:
        # 调用 LLM 生成代码
        return "def hello():\n    print('Hello')"
    
    def fix(self, code: str, errors: List[str]) -> str:
        # 修复代码
        return code


class TesterAgent:
    """测试工程师 Agent"""
    def test(self, code: str) -> Dict:
        # 运行测试
        return {"passed": True, "errors": []}


class ReviewerAgent:
    """代码审查 Agent"""
    def review(self, code: str) -> str:
        # 代码审查
        return "Code looks good!"


# 使用
system = MultiAgentSystem()
result = system.execute_project("创建一个 Hello World 程序")

---

技术 2：Self-Reflection（自我反思）

原理：

传统 Agent：
  执行 → 输出

Self-Reflection Agent：
  执行 → 反思（我做得好吗？） → 改进 → 输出

代码实现：

class SelfReflectingAgent:
    """自我反思 Agent"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def solve_with_reflection(self, problem: str) -> str:
        """带反思的求解"""
        # 第一次尝试
        initial_solution = self.solve(problem)
        
        # 反思
        reflection = self.reflect(problem, initial_solution)
        
        if reflection['needs_improvement']:
            # 改进
            improved_solution = self.improve(
                problem, 
                initial_solution, 
                reflection['feedback']
            )
            return improved_solution
        
        return initial_solution
    
    def solve(self, problem: str) -> str:
        """求解"""
        prompt = f"Solve this problem: {problem}"
        return self.llm.generate(prompt)
    
    def reflect(self, problem: str, solution: str) -> Dict:
        """反思"""
        prompt = f"""
Problem: {problem}

Solution: {solution}

Critique this solution:
1. Is it correct?
2. Is it efficient?
3. Are there edge cases missed?
4. How can it be improved?

Output JSON:
{{
  "is_correct": true/false,
  "needs_improvement": true/false,
  "feedback": "improvement suggestions"
}}
"""
        
        response = self.llm.generate(prompt)
        return json.loads(response)
    
    def improve(self, problem: str, solution: str, feedback: str) -> str:
        """改进"""
        prompt = f"""
Problem: {problem}

Initial Solution: {solution}

Feedback: {feedback}

Provide an improved solution addressing the feedback.
"""
        
        return self.llm.generate(prompt)

---

📊 Agent 能力边界总结

能做的 ✅

能力	示例	成熟度
自动化工作流	数据报告、邮件发送	⭐⭐⭐⭐⭐
多步推理	数学计算、逻辑推理	⭐⭐⭐⭐
工具调用	API 调用、数据库查询	⭐⭐⭐⭐⭐
自主学习	客服机器人积累知识	⭐⭐⭐⭐
代码生成	补全、重构、调试	⭐⭐⭐⭐

不能做的 ❌

限制	原因	缓解方案
真正理解	只是模式匹配	人类监督
创造性思维	基于已有数据	人类创意输入
长链条任务	错误累积	拆解小任务
安全性保证	可能被注入攻击	沙箱、白名单
实时决策	依赖训练数据	结合规则引擎

---

💡 实战建议

1. 何时使用 Agent？

适合场景：
✅ 重复性工作流（自动化）
✅ 多步骤任务（规划能力强）
✅ 需要调用外部工具（API、数据库）
✅ 容错率高（可以重试）

不适合场景：
❌ 高安全性要求（金融交易）
❌ 实时性要求极高（高频交易）
❌ 需要真正创新（艺术创作）
❌ 成本敏感（Agent 比 LLM 贵 5-10 倍）

2. 最佳实践

# 1. 从小任务开始
# 不要一开始就搞复杂的多 Agent 系统

# 2. 完善的错误处理
try:
    result = agent.execute(task)
except Exception as e:
    log_error(e)
    notify_human()

# 3. 监控和日志
logger.info(f"Agent executed: {task}")
logger.info(f"Result: {result}")

# 4. 人类在环（Human-in-the-loop）
if confidence < 0.8:
    ask_human_for_approval()

---

🎯 总结

Agent 核心技术栈：

基础层：
  - LLM（大脑）
  - Memory（记忆）
  - Tools（工具）
  - Planning（规划）

框架层：
  - ReAct（推理 + 行动）
  - Function Calling（工具调用）
  - Tree of Thoughts（思维树）

进阶层：
  - Multi-Agent（多智能体协作）
  - Self-Reflection（自我反思）
  - Learning（持续学习）

安全层：
  - Prompt 注入检测
  - 命令白名单
  - 沙箱执行
  - 超时控制

未来展望：

1. 更自主 - 减少人类干预
2. 更高效 - 降低推理成本
3. 更安全 - 形式化验证
4. 更通用 - 跨领域应用

---

📚 延伸阅读

完整代码和实战案例： https://github.com/Lee985-cmd/AI-30-Day-Challenge

30 天 AI 挑战教程： https://blog.csdn.net/m0_67081842

评论区留言： 你想用 Agent 做什么？

• 自动化工作流？
• 智能客服？
• 其他创意？

欢迎分享你的想法！

---

⭐ 如果这篇文章帮到你了，请 Star GitHub 项目支持一下！ 🌟 Star 链接

已有 6 人 Star，你的支持是我持续更新的动力！