最近在做一个内部项目，需求很简单：用户问一句话，AI 自己判断要不要查天气、搜数据库、调计算器，然后把结果整合起来回复。听着简单对吧？但从"能跑"到"能用"，中间踩了不少坑。

这篇文章把我的完整实现过程分享出来，从基础概念到可运行的代码，适合有 Python 基础、想上手 AI Agent 开发的同学。

什么是 AI Agent？跟普通聊天机器人有啥区别？

先别急着写代码，搞清楚概念很重要。

普通聊天机器人：你问什么，它答什么。输入 → 模型 → 输出，一条线走到底。

AI Agent：你给它一个目标，它自己规划步骤、调用工具、根据中间结果调整策略，最终完成任务。核心区别是——它会"思考"下一步该干什么。

举个例子：

• 聊天机器人：“帮我查一下北京明天的天气” → 回一段天气描述（可能是编的）
• AI Agent：“帮我查一下北京明天的天气” → 调用天气 API → 拿到真实数据 → 格式化返回

Agent 的关键能力有三个：

1. 规划（Planning）：拆解任务，决定执行顺序
2. 工具调用（Tool Use）：调 API、查数据库、执行代码
3. 反思（Reflection）：根据执行结果判断是否需要调整

技术选型

我选的方案：

• Python 3.11+
• OpenAI SDK（兼容各种模型的 Function Calling）
• Claude / GPT 系列模型（都支持原生 Tool Use）
• 不用 LangChain（对于这个规模的项目，直接调 SDK 更可控）

为什么不用 LangChain？不是说它不好，是对于 Agent 入门来说，直接用 SDK 能让你更清楚每一步在干什么。等你理解了底层原理，再上框架事半功倍。

第一步：定义工具（Tools）

Agent 的"手脚"就是工具。我们先定义三个常用的：

import json
import httpx
from datetime import datetime

# 工具 1：获取当前时间
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取指定时区的当前时间"""
    from zoneinfo import ZoneInfo
    now = datetime.now(ZoneInfo(timezone))
    return now.strftime("%Y-%m-%d %H:%M:%S %Z")

# 工具 2：天气查询（用免费 API）
def get_weather(city: str) -> str:
    """查询城市天气"""
    # 这里用 wttr.in 的免费 API
    try:
        resp = httpx.get(f"https://wttr.in/{city}?format=j1", timeout=10)
        data = resp.json()
        current = data["current_condition"][0]
        return json.dumps({
            "city": city,
            "temperature": f"{current['temp_C']}°C",
            "feels_like": f"{current['FeelsLikeC']}°C",
            "description": current["weatherDesc"][0]["value"],
            "humidity": f"{current['humidity']}%"
        }, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": f"查询失败: {str(e)}"})

# 工具 3：简单计算器
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        # 安全限制：只允许数学运算
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return json.dumps({"error": "不支持的字符"})
        result = eval(expression)  # 生产环境建议用 ast.literal_eval 或 sympy
        return json.dumps({"expression": expression, "result": result})
    except Exception as e:
        return json.dumps({"error": f"计算失败: {str(e)}"})

几个注意点：

• 每个工具函数都返回 JSON 字符串，方便模型解析
• 异常处理要做好，Agent 调用失败不能直接崩
• calculate 里用了 eval，生产环境一定要换成安全的实现

第二步：描述工具（让模型知道有哪些工具可用）

模型不会自动知道你有什么工具，你得用 JSON Schema 格式告诉它：

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取指定时区的当前时间。当用户询问时间相关问题时调用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "时区名称，如 Asia/Shanghai、America/New_York",
                        "default": "Asia/Shanghai"
                    }
                },
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气信息，包括温度、体感温度、天气描述和湿度。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，支持中文和英文，如 Beijing、上海"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "计算数学表达式，支持加减乘除和括号。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 (10 + 20) * 3"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

关键经验：description 写得好不好，直接决定模型调工具准不准。写清楚"什么时候该调用"比"这个工具是什么"更重要。

第三步：实现 Agent 核心循环

Agent 的核心就是一个循环：发消息 → 模型决策 → 如果要调工具就调 → 把结果喂回模型 → 重复直到模型觉得可以直接回答了。

from openai import OpenAI

# 初始化客户端
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.ofox.ai/v1"  # 国内直连，支持 Claude/GPT 等多模型
)

# 工具名称到函数的映射
TOOL_MAP = {
    "get_current_time": get_current_time,
    "get_weather": get_weather,
    "calculate": calculate,
}

def run_agent(user_message: str, model: str = "claude-sonnet-4-20250514"):
    """运行 Agent，支持多轮工具调用"""
    
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个有用的助手，可以使用工具来帮助用户。"
                "如果需要查询实时信息（时间、天气等），请调用相应工具。"
                "如果不需要工具就能回答，直接回答即可。"
            )
        },
        {"role": "user", "content": user_message}
    ]
    
    max_rounds = 10  # 防止无限循环
    
    for round_num in range(max_rounds):
        print(f"\n--- Round {round_num + 1} ---")
        
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto"  # 让模型自己决定要不要调工具
        )
        
        msg = response.choices[0].message
        messages.append(msg)  # 把模型回复加到对话历史
        
        # 如果模型决定不调工具，直接返回
        if not msg.tool_calls:
            print(f"Agent 最终回复: {msg.content}")
            return msg.content
        
        # 执行所有工具调用
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)
            
            print(f"调用工具: {func_name}({func_args})")
            
            # 执行工具
            if func_name in TOOL_MAP:
                result = TOOL_MAP[func_name](**func_args)
            else:
                result = json.dumps({"error": f"未知工具: {func_name}"})
            
            print(f"工具返回: {result}")
            
            # 把工具结果加到对话历史
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
    
    return "Agent 达到最大轮次限制，请简化问题后重试。"

这段代码是整个 Agent 的核心，几个关键点：

1. tool_choice="auto"：让模型自己判断要不要调工具，也可以设成 "required" 强制调用
2. 循环执行：模型可能需要多轮工具调用才能完成任务
3. max_rounds 限制：防止模型陷入死循环，生产环境必备
4. 完整的消息历史：每一轮的工具调用和结果都要加到 messages 里，模型需要这些上下文

第四步：测试一下

if __name__ == "__main__":
    # 测试 1：简单问题，不需要工具
    print("=" * 50)
    run_agent("Python 的 GIL 是什么？")
    
    # 测试 2：需要调用一个工具
    print("=" * 50)
    run_agent("现在北京几点了？")
    
    # 测试 3：需要调用多个工具
    print("=" * 50)
    run_agent("帮我查一下上海和东京的天气，然后算一下温差")
    
    # 测试 4：复杂多步任务
    print("=" * 50)
    run_agent("现在几点了？如果是工作时间（9-18点），帮我查下北京天气")

运行测试 3 的输出大概是这样的：

--- Round 1 ---
调用工具: get_weather({"city": "上海"})
工具返回: {"city": "上海", "temperature": "18°C", ...}
调用工具: get_weather({"city": "东京"})
工具返回: {"city": "东京", "temperature": "14°C", ...}

--- Round 2 ---
调用工具: calculate({"expression": "18 - 14"})
工具返回: {"expression": "18 - 14", "result": 4}

--- Round 3 ---
Agent 最终回复: 查询结果如下：
- 上海当前气温 18°C
- 东京当前气温 14°C  
- 温差为 4°C，上海比东京暖和一些。

可以看到模型自己规划了三步：先查两个城市的天气，再用计算器算温差，最后整合结果回复。这就是 Agent 的魅力。

进阶：添加错误重试和流式输出

基础版跑通之后，生产环境还需要几个改进：

import time
from openai import OpenAI, APITimeoutError, APIConnectionError

client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.ofox.ai/v1"
)

def run_agent_v2(user_message: str, model: str = "claude-sonnet-4-20250514"):
    """带重试和流式输出的 Agent"""
    
    messages = [
        {"role": "system", "content": "你是一个有用的助手，可以使用工具来帮助用户。"},
        {"role": "user", "content": user_message}
    ]
    
    max_rounds = 10
    max_retries = 3
    
    for round_num in range(max_rounds):
        # 带重试的 API 调用
        for attempt in range(max_retries):
            try:
                response = client.chat.completions.create(
                    model=model,
                    messages=messages,
                    tools=TOOLS,
                    tool_choice="auto",
                    timeout=30
                )
                break
            except (APITimeoutError, APIConnectionError) as e:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # 指数退避
                    print(f"请求失败，{wait_time}秒后重试: {e}")
                    time.sleep(wait_time)
                else:
                    raise
        
        msg = response.choices[0].message
        messages.append(msg)
        
        if not msg.tool_calls:
            return msg.content
        
        # 并行执行工具调用（如果互不依赖）
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            try:
                func_args = json.loads(tool_call.function.arguments)
                result = TOOL_MAP.get(func_name, lambda **_: '{"error": "未知工具"}')(**func_args)
            except Exception as e:
                result = json.dumps({"error": f"工具执行失败: {str(e)}"})
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })
    
    return "达到最大轮次限制"

改进点：

• 指数退避重试：网络不稳定时自动重试，不会直接挂掉
• 统一的异常处理：工具执行失败也不会中断 Agent 循环
• 超时控制：单次 API 调用 30 秒超时

踩坑记录

分享几个我实际开发中遇到的问题：

1. 工具描述写得太模糊，模型乱调

反面："description": "获取天气" — 模型可能在用户问"天气怎么样？"（闲聊）时也去调 API。

正面："description": "查询指定城市的实时天气。当用户明确询问某个城市的天气状况时调用。" — 加上"什么时候该调用"的说明。

2. 忘了处理模型并行调用工具

Claude 和 GPT 都支持一次返回多个 tool_calls。如果你的代码只处理第一个，就会丢数据。一定要遍历 msg.tool_calls 列表。

3. 消息历史越来越长，Token 爆了

Agent 跑多轮之后，messages 会很长。两个方案：

• 设置 max_rounds 限制轮次
• 超过一定长度时，只保留 system message + 最近 N 轮

4. eval() 安全问题

计算器那个 eval() 在生产环境千万别直接用。可以用 ast.literal_eval 处理简单表达式，或者用 sympy 库做安全计算。

完整项目结构

ai-agent/
├── agent.py          # Agent 核心逻辑
├── tools/
│   ├── __init__.py
│   ├── time_tool.py  # 时间工具
│   ├── weather.py    # 天气工具
│   └── calculator.py # 计算器
├── config.py         # 配置（API Key、模型选择等）
├── requirements.txt
└── main.py           # 入口

requirements.txt：

openai>=1.30.0
httpx>=0.27.0

下一步可以做什么

这篇文章搭了一个基础 Agent 框架，你可以在这个基础上扩展：

1. 加更多工具：搜索引擎、数据库查询、文件读写、代码执行
2. 加记忆：用 SQLite 或 Redis 存对话历史，让 Agent 有长期记忆
3. 接入 MCP：用 Model Context Protocol 标准化工具接入，可以一次对接大量工具
4. 多 Agent 协作：让多个专业 Agent 各司其职，通过消息传递协作

AI Agent 开发没有想象中那么难，核心就是工具定义 + 循环调用 + 结果整合。理解了这个模式，剩下的就是往里面填工具和优化体验。

希望这篇对你有帮助，有问题欢迎评论区讨论。