Python调用OpenAI Agents SDK打造一个多智能体系统

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
# 安装 Agents SDK
pip install openai-agents
# 验证安装
python -c "import agents; print(agents.__version__)"
# 输出: 0.14.2

import os
# 方式1: 环境变量
os.environ["OPENAI_API_KEY"] = "sk-your-api-key-here"
# 方式2: .env 文件（推荐）
# 在项目根目录创建 .env 文件:
# OPENAI_API_KEY=sk-your-api-key-here

import asyncio
from agents import Agent, Runner

async def main():
    # 创建一个Agent —— 就这么简单
    agent = Agent(
        name="Haiku Master",
        instructions="你只使用俳句格式回答问题。俳句是5-7-5音节的三行诗。",
    )

    # 运行Agent
    result = await Runner.run(agent, "用Python写递归是什么感觉？")
    print(result.final_output)
    # 输出:
    # 函数调自身，
    # 分解问题为小块，
    # 无尽由设计。

if __name__ == "__main__":
    asyncio.run(main)

from pydantic import BaseModel

class ShoppingContext(BaseModel):
    """购物助手的共享上下文"""
    customer_id: str | None = None
    customer_name: str | None = None
    order_id: str | None = None
    cart_items: list[str] = []
    is_premium_user: bool = False

import random
from agents import RunContextWrapper, function_tool

# ===== 商品相关工具 =====

@function_tool
async def search_products(query: str) -> str:
    """搜索商品。query是搜索关键词。"""
    # 实际项目中这里连接搜索引擎或数据库
    mock_products = {
        "手机": "iPhone 16 Pro (¥8,999), Samsung S26 (¥6,999), Xiaomi 16 (¥3,499)",
        "笔记本": "MacBook Pro M5 (¥14,999), ThinkPad X1 (¥9,999), Surface Laptop (¥8,999)",
        "耳机": "AirPods Pro 3 (¥1,899), Sony WH-1000XM6 (¥2,499), Bose QC Ultra (¥2,299)",
    }
    for keyword, products in mock_products.items():
        if keyword in query:
            return f"找到以下商品: {products}"
    return f"未找到与 '{query}' 相关的商品，试试其他关键词？"


@function_tool
async def get_product_detail(product_name: str) -> str:
    """获取商品详情。product_name是商品名称。"""
    details = {
        "iPhone 16 Pro": "A20芯片 | 6.3英寸 | 4800万像素 | 256GB起 | 电池续航30小时",
        "MacBook Pro M5": "M5芯片 | 16英寸 | 36GB内存 | 512GB SSD | 续航24小时",
    }
    return details.get(product_name, f"商品 '{product_name}' 详情: 评分4.8 | 月销10000+ | 好评率98%")


@function_tool
async def add_to_cart(
    context: RunContextWrapper[ShoppingContext],
    product_name: str,
    quantity: int = 1
) -> str:
    """将商品加入购物车。"""
    context.context.cart_items.append(f"{product_name} x{quantity}")
    return (f"已将 {product_name} x{quantity} 加入购物车。"
            f"当前购物车: {', '.join(context.context.cart_items)}")


# ===== 订单相关工具 =====

@function_tool
async def check_order_status(
    context: RunContextWrapper[ShoppingContext],
    order_id: str
) -> str:
    """查询订单状态。order_id是订单号。"""
    context.context.order_id = order_id
    statuses = ["已发货", "运输中", "派送中", "已签收"]
    status = random.choice(statuses)
    return f"订单 {order_id} 状态: {status}。预计到达: 明天下午。"


@function_tool
async def process_refund(
    context: RunContextWrapper[ShoppingContext],
    order_id: str,
    reason: str
) -> str:
    """处理退款。order_id是订单号，reason是退款原因。"""
    if context.context.is_premium_user:
        return f"VIP用户优先退款已提交！订单 {order_id}，原因: {reason}。退款将在1-2个工作日内到账。"
    return f"退款已提交！订单 {order_id}，原因: {reason}。退款将在3-5个工作日内到账。"


@function_tool
async def track_package(order_id: str) -> str:
    """追踪包裹物流。"""
    locations = ["北京分拣中心", "上海转运站", "杭州配送站", "客户所在城市"]
    location = random.choice(locations)
    return f"订单 {order_id} 最新位置: {location}。更新时间: 10分钟前。"

from pydantic import BaseModel, Field
from agents import (
    GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
    RunContextWrapper, TResponseInputItem, Agent, Runner,
    input_guardrail,
)

# ===== 输入护栏: 检查是否为购物相关问题 =====

class RelevanceCheck(BaseModel):
    is_shopping_related: bool = Field(description="是否与购物/商品/订单相关")
    reasoning: str = Field(description="判断理由")

guardrail_agent = Agent(
    name="Relevance Guard",
    instructions="""判断用户输入是否与以下领域相关:
    - 商品搜索与推荐
    - 订单查询与物流
    - 退款与售后
    - 购物咨询
    如果完全无关，标记为不相关。""",
    output_type=RelevanceCheck,
)

@input_guardrail
async def shopping_relevance_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    """确保Agent只处理购物相关问题"""
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    check = result.final_output_as(RelevanceCheck)
    return GuardrailFunctionOutput(
        output_info=check.model_dump(),
        tripwire_triggered=not check.is_shopping_related,
    )

from agents import Agent, handoff
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX

# ===== Agent 1: 商品专家 =====

product_agent = Agent[ShoppingContext](
    name="Product Specialist",
    handoff_description="处理商品搜索、推荐、详情查询和加购。",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是一个热情的商品专家，名叫小商。

## 你的职责
- 帮助用户搜索和推荐商品
- 提供商品详细信息
- 协助用户将商品加入购物车

## 工作流程
1. 理解用户的购物需求
2. 使用 search_products 搜索相关商品
3. 如果用户想了解某个商品，使用 get_product_detail
4. 如果用户决定购买，使用 add_to_cart

## 规则
- 推荐时优先考虑性价比
- 如果用户的需求超出了商品范围（如查订单、退款），交接给订单专家
- 保持友好专业的语气
""",
    tools=[search_products, get_product_detail, add_to_cart],
)

# ===== Agent 2: 订单专家 =====

order_agent = Agent[ShoppingContext](
    name="Order Specialist",
    handoff_description="处理订单查询、物流追踪和退款处理。",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是一个耐心的订单专家，名叫小单。

## 你的职责
- 查询订单状态
- 追踪包裹物流
- 处理退款请求

## 工作流程
1. 确认用户的订单号（如果没有，先询问）
2. 根据需求使用对应工具:
   - 查状态 → check_order_status
   - 查物流 → track_package
   - 退款 → process_refund

## 规则
- 处理退款时务必确认订单号和退款原因
- VIP用户享有优先退款
- 如果用户想继续购物，交接给商品专家
- 保持耐心和同理心
""",
    tools=[check_order_status, track_package, process_refund],
)

# ===== Agent 3: 分诊路由 =====

triage_agent = Agent[ShoppingContext](
    name="Shopping Triage",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是一个智能购物助手的接待员。

## 你的职责
判断用户的需求类型，将其路由到正确的专家:
- 商品搜索/推荐/购买 → 商品专家 (Product Specialist)
- 订单查询/物流/退款 → 订单专家 (Order Specialist)

## 规则
- 简单打招呼由你自己回复
- 无法判断时，优先交给商品专家
- 用一句话简单介绍自己后立即路由
""",
    input_guardrails=[shopping_relevance_guardrail],
    handoffs=[
        handoff(agent=product_agent),
        handoff(agent=order_agent),
    ],
)

# 允许专家之间互相交接
product_agent.handoffs.append(handoff(agent=order_agent))
order_agent.handoffs.append(handoff(agent=product_agent))
# 允许专家回到分诊
product_agent.handoffs.append(handoff(agent=triage_agent))
order_agent.handoffs.append(handoff(agent=triage_agent))

import uuid
from agents import (
    Runner, trace, HandoffOutputItem, ItemHelpers,
    MessageOutputItem, ToolCallItem, ToolCallOutputItem,
)

async def main():
    current_agent = triage_agent
    input_items: list = []
    context = ShoppingContext(is_premium_user=True)  # 模拟VIP用户
    conversation_id = uuid.uuid4().hex[:16]

    print("=" * 60)
    print("   欢迎使用智能购物助手!")
    print("   输入 'quit' / 'exit' / 'bye' 退出")
    print("=" * 60)

    while True:
        user_input = input("\n🧑 你: ").strip()
        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit", "bye"):
            print("👋 感谢使用，再见！")
            break

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

        try:
            with trace("Shopping Assistant", group_id=conversation_id):
                result = await Runner.run(
                    current_agent,
                    input_items,
                    context=context,
                )

                # 打印详细的执行过程
                for new_item in result.new_items:
                    agent_name = new_item.agent.name

                    if isinstance(new_item, MessageOutputItem):
                        text = ItemHelpers.text_message_output(new_item)
                        if text:
                            print(f"\n🤖 [{agent_name}]: {text}")

                    elif isinstance(new_item, HandoffOutputItem):
                        print(f"  🔄 [交接] {new_item.source_agent.name} "
                              f"→ {new_item.target_agent.name}")

                    elif isinstance(new_item, ToolCallItem):
                        print(f"  🔧 [{agent_name}]: 正在调用工具...")

                    elif isinstance(new_item, ToolCallOutputItem):
                        print(f"  📋 [工具结果]: {str(new_item.output)[:100]}...")

                input_items = result.to_input_list()
                current_agent = result.last_agent

        except InputGuardrailTripwireTriggered as e:
            message = "抱歉，我只能帮助处理购物相关问题（商品、订单、退款等）。请换个问题试试？"
            print(f"\n🛡️ [护栏触发]: {message}")
            input_items.append({"role": "assistant", "content": message})


if __name__ == "__main__":
    asyncio.run(main())

============================================================
   欢迎使用智能购物助手!
   输入 'quit' / 'exit' / 'bye' 退出
============================================================

🧑 你: 你好，我想买个手机

🤖 [Shopping Triage]: 你好！我是智能购物助手，马上为你找商品专家！
  🔄 [交接] Shopping Triage → Product Specialist

🤖 [Product Specialist]: 你好！我是商品专家小商 🎉 来帮你挑手机！
  🔧 [Product Specialist]: 正在调用工具...
  📋 [工具结果]: 找到以下商品: iPhone 16 Pro (¥8,999), Samsung S26 (¥6,999), Xiaomi 16 (¥3,499)...

🤖 [Product Specialist]: 我为你找到了几款热门手机：
- iPhone 16 Pro - ¥8,999
- Samsung S26 - ¥6,999
- Xiaomi 16 - ¥3,499

你预算大概多少？我可以进一步推荐！

🧑 你: iPhone 16 Pro 怎么样？

  🔧 [Product Specialist]: 正在调用工具...
  📋 [工具结果]: A20芯片 | 6.3英寸 | 4800万像素 | 256GB起 | 电池续航30小时

🤖 [Product Specialist]: iPhone 16 Pro 是今年的旗舰款，配置非常强：
- A20芯片，性能提升40%
- 4800万像素主摄
- 电池续航长达30小时

要加入购物车吗？

🧑 你: 好的，加一辆

  🔧 [Product Specialist]: 正在调用工具...
  📋 [工具结果]: 已将 iPhone 16 Pro x1 加入购物车...
🤖 [Product Specialist]: 已加入购物车！还需要别的吗？或者我可以帮你查一下订单？

🧑 你: 帮我查一下我之前的订单 20260315001

  🔄 [交接] Product Specialist → Order Specialist
  🔧 [Order Specialist]: 正在调用工具...
  📋 [工具结果]: 订单 20260315001 状态: 运输中。预计到达: 明天下午。

🤖 [Order Specialist]: 你的订单 20260315001 正在运输中，预计明天下午就能收到啦！

🧑 你: 天气怎么样？

🛡️ [护栏触发]: 抱歉，我只能帮助处理购物相关问题（商品、订单、退款等）。请换个问题试试？

# ===== Agents as Tools 模式 =====
# 场景: 内容创作流水线 —— 一个Manager Agent协调调研、写作、审核

from agents import Agent, Runner, trace

# 专家Agent
researcher = Agent(
    name="Researcher",
    instructions="你是一个调研专家。对给定主题进行深度调研，输出结构化的调研报告。",
)

writer = Agent(
    name="Writer",
    instructions="你是一个资深技术写手。基于调研报告撰写高质量的技术文章。",
)

reviewer = Agent(
    name="Reviewer",
    instructions="""你是一个严格的审核编辑。从以下维度审核文章:
    1. 技术准确性 (0-10)
    2. 可读性 (0-10)
    3. 完整性 (0-10)
    输出JSON格式的评分和修改建议。""",
    output_type=dict,  # 结构化输出
)

# Manager Agent —— 把专家当作工具使用
manager = Agent(
    name="Content Manager",
    instructions="""你是一个内容管理Agent，负责协调一篇文章的创作。
    工作流程:
    1. 先用 research 工具调研主题
    2. 再用 write 工具基于调研结果撰写文章
    3. 最后用 review 工具审核文章质量
    4. 如果评分低于8分，根据意见修改后重新审核
    """,
    tools=[
        researcher.as_tool(
            tool_name="research",
            tool_description="对指定主题进行深度调研，返回调研报告"
        ),
        writer.as_tool(
            tool_name="write",
            tool_description="基于调研报告撰写技术文章"
        ),
        reviewer.as_tool(
            tool_name="review",
            tool_description="审核文章质量，返回评分和修改建议"
        ),
    ],
)

async def create_article(topic: str) -> str:
    with trace("Content Creation Pipeline"):
        result = await Runner.run(
            manager,
            f"请创作一篇关于 '{topic}' 的技术文章"
        )
    return result.final_output

from agents import Agent, Runner, trace

# 方式1: 自动Tracing（默认开启）
result = await Runner.run(agent, "Hello")
# 自动在 OpenAI Dashboard 中创建 trace

# 方式2: 手动创建 trace（跨多个 run）
with trace("复杂工作流", group_id="session-123"):
    result1 = await Runner.run(agent1, "步骤1")
    result2 = await Runner.run(agent2, f"基于上一步: {result1.final_output}")
    result3 = await Runner.run(agent3, f"最终处理: {result2.final_output}")

# 方式3: 关闭 Tracing
from agents import set_tracing_disabled
set_tracing_disabled(True)

# 集成 LangSmith / Weights & Biases / MLflow 等
from agents import add_trace_processor

# 方式1: 添加自定义处理器（保留OpenAI后端）
add_trace_processor(my_custom_processor)

# 方式2: 完全替换处理器
from agents import set_trace_processors
set_trace_processors([langsmith_processor, wandb_processor])

from agents import Runner, trace, flush_traces

async def background_agent_task(prompt: str):
    try:
        with trace("background_task"):
            result = await Runner.run(agent, prompt)
        return result.final_output
    finally:
        # 确保在Celery/FastAPI等场景中刷新trace
        flush_traces()