使用Python从零搭建一个能用的AI Agent

pip install openai

from openai import OpenAI

client = OpenAI(
    api_key="your-key",
    base_url="https://api.ofox.ai/v1"  # 聚合接口，一个 Key 用所有模型
)

# 模拟的业务函数
def query_order(order_id: str) -> dict:
    """根据订单号查询订单信息"""
    # 实际项目里这里查数据库
    fake_db = {
        "ORD001": {"order_id": "ORD001", "product": "机械键盘", "status": "已发货", "amount": 399},
        "ORD002": {"order_id": "ORD002", "product": "显示器支架", "status": "待付款", "amount": 89},
    }
    return fake_db.get(order_id, {"error": f"订单 {order_id} 不存在"})


def query_logistics(order_id: str) -> dict:
    """根据订单号查询物流信息"""
    fake_logistics = {
        "ORD001": {"carrier": "顺丰", "tracking_no": "SF1234567890", "status": "在途中，预计明天到"},
    }
    return fake_logistics.get(order_id, {"error": f"订单 {order_id} 暂无物流信息"})

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "根据订单号查询订单详情，包括商品名、状态、金额",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单编号，格式如 ORD001"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_logistics",
            "description": "根据订单号查询物流状态，包括快递公司、运单号、当前状态",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单编号，格式如 ORD001"
                    }
                },
                "required": ["order_id"]
            }
        }
    }
]

import json

# 工具名 → 实际函数的映射
TOOL_MAP = {
    "query_order": query_order,
    "query_logistics": query_logistics,
}

SYSTEM_PROMPT = """你是一个电商客服助手。你可以帮用户查询订单信息和物流状态。
请用简洁友好的语气回复。如果用户没有提供订单号，请先询问订单号。"""


def run_agent(user_input: str, messages: list = None, max_turns: int = 5) -> str:
    """
    运行 Agent 主循环
    max_turns: 最大工具调用轮次，防止死循环
    """
    if messages is None:
        messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    messages.append({"role": "user", "content": user_input})

    for turn in range(max_turns):
        response = client.chat.completions.create(
            model="gpt-4o",  # 换成 claude-3.5-sonnet 等也行
            messages=messages,
            tools=tools,
            tool_choice="auto",  # 让模型自己决定要不要调工具
        )

        msg = response.choices[0].message
        messages.append(msg)  # 把 assistant 的回复加入上下文

        # 如果没有工具调用，说明模型准备好直接回答了
        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)

            print(f"  [Agent] 调用工具: {func_name}({func_args})")

            # 执行函数
            if func_name in TOOL_MAP:
                result = TOOL_MAP[func_name](**func_args)
            else:
                result = {"error": f"未知工具: {func_name}"}

            # 把工具执行结果塞回消息列表
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

    return "抱歉，我处理这个问题遇到了困难，请联系人工客服。"

if __name__ == "__main__":
    # 测试 1：正常查询
    print("=" * 50)
    print("用户：帮我看看 ORD001 到哪了")
    print("Agent：", run_agent("帮我看看 ORD001 到哪了"))

    print()

    # 测试 2：需要先查订单再查物流
    print("=" * 50)
    print("用户：ORD001 买的什么？快递到哪了？")
    print("Agent：", run_agent("ORD001 买的什么？快递到哪了？"))

    print()

    # 测试 3：缺少订单号
    print("=" * 50)
    print("用户：我想查一下我的快递")
    print("Agent：", run_agent("我想查一下我的快递"))

try:
    func_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
    # 尝试用 ast.literal_eval 兜底，再不行就报错
    import ast
    try:
        func_args = ast.literal_eval(tool_call.function.arguments)
    except:
        result = {"error": "参数解析失败"}
        # 继续把 error 喂给模型，让它重试

def trim_messages(messages: list, max_tokens: int = 8000) -> list:
    """保留 system prompt + 最近的消息"""
    system_msg = messages[0]  # system prompt 永远保留
    recent = messages[1:]

    # 粗略估算：1个中文字符约2个token
    while len(json.dumps(recent, ensure_ascii=False)) > max_tokens * 2 and len(recent) > 2:
        recent.pop(0)

    return [system_msg] + recent