Python搭AI Agent实战指南,太实用了!小白上手也能做!

用户输入
↓
模型推理（我需要调工具吗？）
├── 不需要 → 直接输出最终回复
└── 需要 → 输出工具调用指令（工具名 + 参数）
↓
你写的Python代码执行工具
↓
把执行结果塞回对话历史
↓
模型继续推理（可能再调一次工具，也可能输出最终答案）
↓
最终回复用户

pip install openai python-dotenv

import os
import json
from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 工具定义：告诉模型有什么工具可以用
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息，包括温度、天气状况和湿度。适用于用户询问天气、是否适合出行等场景。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，使用中文，如：北京、上海、深圳"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认摄氏度"
}
},
"required": ["city"]
}
}
}
]
# 工具实现：真正去执行的函数
def get_weather(city: str, unit: str = "celsius") -> str:
# 真实项目里这里接天气API（如和风天气、心知天气）
# 这里用模拟数据演示
weather_data = {
"北京": {"temp": 22, "condition": "晴天", "humidity": "45%", "advice": "适合出行，无需带雨具"},
"上海": {"temp": 26, "condition": "多云", "humidity": "70%", "advice": "稍热，建议带水"},
"广州": {"temp": 30, "condition": "小雨", "humidity": "85%", "advice": "记得带伞"},
"成都": {"temp": 18, "condition": "阴天", "humidity": "60%", "advice": "温度舒适，轻薄外套即可"},
}
data = weather_data.get(city, {"temp": 20, "condition": "晴天", "humidity": "50%", "advice": "天气不错"})
temp = data["temp"] if unit == "celsius" else round(data["temp"] * 9/5 + 32, 1)
unit_label = "°C" if unit == "celsius" else "°F"
return json.dumps({
"city": city,
"temperature": f"{temp}{unit_label}",
"condition": data["condition"],
"humidity": data["humidity"],
"advice": data["advice"]
}, ensure_ascii=False)
# 工具映射：函数名 → 函数对象
TOOL_FUNCTIONS = {"get_weather": get_weather}
# Agent主循环
def run_agent(user_message: str):
messages = [{"role": "user", "content": user_message}]
while True:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice="auto"
)
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
func_args = json.loads(tool_call.function.arguments)
result = TOOL_FUNCTIONS[func_name](**func_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 测试
print(run_agent("北京今天天气怎么样？我想出去跑步，合适吗？"))

tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，中文"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_products",
"description": "搜索产品数据库，返回产品名称、价格和库存状态。适用于用户询问产品价格、寻找特定商品时。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词，如：MacBook Pro、AirPods"},
"category": {"type": "string", "description": "产品分类，可选：electronics、accessories"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算，返回计算结果。适用于价格合计、折扣计算等数值运算场景。",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如：18999 + 1899，或 18999 * 0.9"
}
},
"required": ["expression"]
}
}
}
]
def search_products(query: str, category: str = None) -> str:
# 真实项目里连接数据库或调商品API
products_db = {
"MacBook Pro 16": {"price": 18999, "stock": "有货", "category": "electronics"},
"MacBook Pro 14": {"price": 14999, "stock": "有货", "category": "electronics"},
"AirPods Pro 3": {"price": 1899, "stock": "有货", "category": "accessories"},
"AirPods 4": {"price": 1099, "stock": "预售", "category": "accessories"},
"iPad Pro 13": {"price": 9999, "stock": "有货", "category": "electronics"},
"Magic Keyboard": {"price": 799, "stock": "有货", "category": "accessories"},
}
results = [
{"name": k, "price": v["price"], "stock": v["stock"]}
for k, v in products_db.items()
if query.lower() in k.lower()
]
if not results:
return json.dumps({"results": [], "message": "未找到匹配产品"}, ensure_ascii=False)
return json.dumps({"results": results, "count": len(results)}, ensure_ascii=False)
def calculate(expression: str) -> str:
try:
# 生产环境用更安全的解析库，如 simpleeval
# 这里为演示用 eval，实际部署需替换
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return json.dumps({"error": "表达式包含不允许的字符"})
result = eval(expression)
return json.dumps({"expression": expression, "result": result, "formatted": f"¥{result:,.2f}"})
except Exception as e:
return json.dumps({"error": f"计算出错: {str(e)}"})
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"search_products": search_products,
"calculate": calculate
}

print(run_agent("帮我找一下MacBook Pro 16和AirPods Pro 3的价格，算算一起买多少钱，打九折的话呢"))

# 权限配置表：每个工具的安全级别
TOOL_PERMISSIONS = {
"get_weather": {
"level": "read",          # 只读，随意调
"rate_limit": 60,         # 每分钟最多60次
"require_confirm": False
},
"search_products": {
"level": "read",
"rate_limit": 30,
"require_confirm": False
},
"calculate": {
"level": "compute",
"rate_limit": 100,
"require_confirm": False
},
"send_email": {
"level": "write",
"rate_limit": 10,
"require_confirm": True,  # 需要人工确认
"reason": "发送邮件操作不可逆，需要确认"
},
"delete_order": {
"level": "dangerous",
"rate_limit": 3,
"require_confirm": True,
"reason": "删除订单数据不可恢复，必须人工审批"
}
}
# 调用计数器（实际项目用Redis等持久化存储）
tool_call_counts = {}
def execute_tool_safely(func_name: str, func_args: dict) -> str:
"""带权限检查的工具执行"""
perm = TOOL_PERMISSIONS.get(func_name)
if not perm:
return json.dumps({"error": "未授权工具，拒绝执行", "tool": func_name})
if func_name not in TOOL_FUNCTIONS:
return json.dumps({"error": f"工具不存在: {func_name}"})
# 频率限制检查
count_key = func_name
tool_call_counts[count_key] = tool_call_counts.get(count_key, 0) + 1
if tool_call_counts[count_key] > perm.get("rate_limit", 999):
return json.dumps({"error": "调用频率超限，请稍后重试"})
# 危险操作：拦截，等待人工确认
if perm.get("require_confirm"):
# 真实项目里这里发通知（钉钉/企微/邮件）给审批人
# 把待审批操作存入数据库，等待回调
return json.dumps({
"status": "pending_approval",
"message": f"此操作需要人工确认：{perm.get('reason', '安全审批')}",
"tool": func_name,
"args": func_args,
"approval_required": True
})
# 正常执行
try:
return TOOL_FUNCTIONS[func_name](func_args)
except TypeError as e:
return json.dumps({"error": f"参数错误: {str(e)}"})
except Exception as e:
return json.dumps({"error": f"工具执行失败: {str(e)}"})

import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
logger = logging.getLogger(__name__)
def run_agent_production(user_message: str, max_rounds: int = 10, verbose: bool = False) -> str:
"""
生产级Agent，带完整错误处理
Args:
user_message: 用户输入
max_rounds: 最大工具调用轮次，防止无限循环
verbose: 是否打印调试信息
"""
messages = [
{
"role": "system",
"content": """你是一个智能助手，可以使用工具查询天气、搜索产品和进行计算。
工具使用规范：
1. 查产品前先确认用户的需求，避免无效搜索
2. 计算前先确认已获取所有需要的数值
3. 如果工具返回错误，告知用户并尝试其他方式解决
4. 如果用户的需求超出工具能力范围，直接说明"""
},
{"role": "user", "content": user_message}
]
total_tokens = 0
start_time = time.time()
for round_num in range(max_rounds):
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice="auto",
timeout=30
)
except Exception as e:
logger.error(f"API调用失败（第{round_num+1}轮）: {e}")
return f"服务暂时不可用，请稍后重试。（错误：{type(e).__name__}）"
# 统计token消耗
if hasattr(response, 'usage') and response.usage:
total_tokens += response.usage.total_tokens
msg = response.choices[0].message
messages.append(msg)
if verbose:
logger.info(f"第{round_num+1}轮推理完成，工具调用：{len(msg.tool_calls) if msg.tool_calls else 0}个")
# 无工具调用，返回最终答案
if not msg.tool_calls:
elapsed = round(time.time() - start_time, 2)
if verbose:
logger.info(f"完成！总轮次：{round_num+1}，总token：{total_tokens}，耗时：{elapsed}s")
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)
except json.JSONDecodeError:
result = json.dumps({"error": "参数解析失败，JSON格式错误"})
func_args = {}
else:
if verbose:
logger.info(f"  调用工具: {func_name}，参数: {func_args}")
result = execute_tool_safely(func_name, func_args)
if verbose:
logger.info(f"  返回结果: {result[:100]}{'...' if len(result) > 100 else ''}")
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
logger.warning(f"达到最大轮次限制（{max_rounds}）")
return f"任务处理超出限制，请将问题拆分后重试。"