# 創(chuàng)建虛擬環(huán)境
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: 環(huán)境變量
os.environ["OPENAI_API_KEY"] = "sk-your-api-key-here"
# 方式2: .env 文件（推薦）
# 在項目根目錄創(chuàng)建 .env 文件:
# OPENAI_API_KEY=sk-your-api-key-here

import asyncio
from agents import Agent, Runner

async def main():
    # 創(chuàng)建一個Agent —— 就這么簡單
    agent = Agent(
        name="Haiku Master",
        instructions="你只使用俳句格式回答問題。俳句是5-7-5音節(jié)的三行詩。",
    )

    # 運行Agent
    result = await Runner.run(agent, "用Python寫遞歸是什么感覺？")
    print(result.final_output)
    # 輸出:
    # 函數(shù)調自身，
    # 分解問題為小塊，
    # 無盡由設計。

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是搜索關鍵詞。"""
    # 實際項目中這里連接搜索引擎或數(shù)據(jù)庫
    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起 | 電池續(xù)航30小時",
        "MacBook Pro M5": "M5芯片 | 16英寸 | 36GB內存 | 512GB SSD | 續(xù)航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:
    """查詢訂單狀態(tài)。order_id是訂單號。"""
    context.context.order_id = order_id
    statuses = ["已發(fā)貨", "運輸中", "派送中", "已簽收"]
    status = random.choice(statuses)
    return f"訂單 {order_id} 狀態(tài): {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用戶優(yōu)先退款已提交！訂單 {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}
你是一個熱情的商品專家，名叫小商。

## 你的職責
- 幫助用戶搜索和推薦商品
- 提供商品詳細信息
- 協(xié)助用戶將商品加入購物車

## 工作流程
1. 理解用戶的購物需求
2. 使用 search_products 搜索相關商品
3. 如果用戶想了解某個商品，使用 get_product_detail
4. 如果用戶決定購買，使用 add_to_cart

## 規(guī)則
- 推薦時優(yōu)先考慮性價比
- 如果用戶的需求超出了商品范圍（如查訂單、退款），交接給訂單專家
- 保持友好專業(yè)的語氣
""",
    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}
你是一個耐心的訂單專家，名叫小單。

## 你的職責
- 查詢訂單狀態(tài)
- 追蹤包裹物流
- 處理退款請求

## 工作流程
1. 確認用戶的訂單號（如果沒有，先詢問）
2. 根據(jù)需求使用對應工具:
   - 查狀態(tài) → check_order_status
   - 查物流 → track_package
   - 退款 → process_refund

## 規(guī)則
- 處理退款時務必確認訂單號和退款原因
- VIP用戶享有優(yōu)先退款
- 如果用戶想繼續(xù)購物，交接給商品專家
- 保持耐心和同理心
""",
    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)

## 規(guī)則
- 簡單打招呼由你自己回復
- 無法判斷時，優(yōu)先交給商品專家
- 用一句話簡單介紹自己后立即路由
""",
    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,
                )

                # 打印詳細的執(zhí)行過程
                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??? [護欄觸發(fā)]: {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起 | 電池續(xù)航30小時

?? [Product Specialist]: iPhone 16 Pro 是今年的旗艦款，配置非常強：
- A20芯片，性能提升40%
- 4800萬像素主攝
- 電池續(xù)航長達30小時

要加入購物車嗎？

?? 你: 好的，加一輛

  ?? [Product Specialist]: 正在調用工具...
  ?? [工具結果]: 已將 iPhone 16 Pro x1 加入購物車...
?? [Product Specialist]: 已加入購物車！還需要別的嗎？或者我可以幫你查一下訂單？

?? 你: 幫我查一下我之前的訂單 20260315001

  ?? [交接] Product Specialist → Order Specialist
  ?? [Order Specialist]: 正在調用工具...
  ?? [工具結果]: 訂單 20260315001 狀態(tài): 運輸中。預計到達: 明天下午。

?? [Order Specialist]: 你的訂單 20260315001 正在運輸中，預計明天下午就能收到啦！

?? 你: 天氣怎么樣？

??? [護欄觸發(fā)]: 抱歉，我只能幫助處理購物相關問題（商品、訂單、退款等）。請換個問題試試？

# ===== Agents as Tools 模式 =====
# 場景: 內容創(chuàng)作流水線 —— 一個Manager Agent協(xié)調調研、寫作、審核

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，負責協(xié)調一篇文章的創(chuàng)作。
    工作流程:
    1. 先用 research 工具調研主題
    2. 再用 write 工具基于調研結果撰寫文章
    3. 最后用 review 工具審核文章質量
    4. 如果評分低于8分，根據(jù)意見修改后重新審核
    """,
    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"請創(chuàng)作一篇關于 '{topic}' 的技術文章"
        )
    return result.final_output

from agents import Agent, Runner, trace

# 方式1: 自動Tracing（默認開啟）
result = await Runner.run(agent, "Hello")
# 自動在 OpenAI Dashboard 中創(chuàng)建 trace

# 方式2: 手動創(chuàng)建 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()