# 創(chuàng)建虛擬環(huán)境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
# 安裝 Agents SDK
pip install openai-agents
# 驗(yàn)證安裝
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 文件（推薦）
# 在項(xiàng)目根目錄創(chuàng)建 .env 文件:
# OPENAI_API_KEY=sk-your-api-key-here

import asyncio
from agents import Agent, Runner

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

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

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

# ===== 商品相關(guān)工具 =====

@function_tool
async def search_products(query: str) -> str:
    """搜索商品。query是搜索關(guān)鍵詞。"""
    # 實(shí)際項(xiàng)目中這里連接搜索引擎或數(shù)據(jù)庫
    mock_products = {
        "手機(jī)": "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)",
        "耳機(jī)": "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}' 相關(guān)的商品，試試其他關(guān)鍵詞？"


@function_tool
async def get_product_detail(product_name: str) -> str:
    """獲取商品詳情。product_name是商品名稱。"""
    details = {
        "iPhone 16 Pro": "A20芯片 | 6.3英寸 | 4800萬像素 | 256GB起 | 電池續(xù)航30小時(shí)",
        "MacBook Pro M5": "M5芯片 | 16英寸 | 36GB內(nèi)存 | 512GB SSD | 續(xù)航24小時(shí)",
    }
    return details.get(product_name, f"商品 '{product_name}' 詳情: 評(píng)分4.8 | 月銷10000+ | 好評(píng)率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"當(dāng)前購物車: {', '.join(context.context.cart_items)}")


# ===== 訂單相關(guān)工具 =====

@function_tool
async def check_order_status(
    context: RunContextWrapper[ShoppingContext],
    order_id: str
) -> str:
    """查詢訂單狀態(tài)。order_id是訂單號(hào)。"""
    context.context.order_id = order_id
    statuses = ["已發(fā)貨", "運(yùn)輸中", "派送中", "已簽收"]
    status = random.choice(statuses)
    return f"訂單 {order_id} 狀態(tài): {status}。預(yù)計(jì)到達(dá): 明天下午。"


@function_tool
async def process_refund(
    context: RunContextWrapper[ShoppingContext],
    order_id: str,
    reason: str
) -> str:
    """處理退款。order_id是訂單號(hào)，reason是退款原因。"""
    if context.context.is_premium_user:
        return f"VIP用戶優(yōu)先退款已提交！訂單 {order_id}，原因: {reason}。退款將在1-2個(gè)工作日內(nèi)到賬。"
    return f"退款已提交！訂單 {order_id}，原因: {reason}。退款將在3-5個(gè)工作日內(nèi)到賬。"


@function_tool
async def track_package(order_id: str) -> str:
    """追蹤包裹物流。"""
    locations = ["北京分揀中心", "上海轉(zhuǎn)運(yùn)站", "杭州配送站", "客戶所在城市"]
    location = random.choice(locations)
    return f"訂單 {order_id} 最新位置: {location}。更新時(shí)間: 10分鐘前。"

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

# ===== 輸入護(hù)欄: 檢查是否為購物相關(guān)問題 =====

class RelevanceCheck(BaseModel):
    is_shopping_related: bool = Field(description="是否與購物/商品/訂單相關(guān)")
    reasoning: str = Field(description="判斷理由")

guardrail_agent = Agent(
    name="Relevance Guard",
    instructions="""判斷用戶輸入是否與以下領(lǐng)域相關(guān):
    - 商品搜索與推薦
    - 訂單查詢與物流
    - 退款與售后
    - 購物咨詢
    如果完全無關(guān)，標(biāo)記為不相關(guān)。""",
    output_type=RelevanceCheck,
)

@input_guardrail
async def shopping_relevance_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    """確保Agent只處理購物相關(guān)問題"""
    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}
你是一個(gè)熱情的商品專家，名叫小商。

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

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

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

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

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

## 規(guī)則
- 處理退款時(shí)務(wù)必確認(rèn)訂單號(hào)和退款原因
- 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}
你是一個(gè)智能購物助手的接待員。

## 你的職責(zé)
判斷用戶的需求類型，將其路由到正確的專家:
- 商品搜索/推薦/購買 → 商品專家 (Product Specialist)
- 訂單查詢/物流/退款 → 訂單專家 (Order Specialist)

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

                # 打印詳細(xì)的執(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}]: 正在調(diào)用工具...")

                    elif isinstance(new_item, ToolCallOutputItem):
                        print(f"  ?? [工具結(jié)果]: {str(new_item.output)[:100]}...")

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

        except InputGuardrailTripwireTriggered as e:
            message = "抱歉，我只能幫助處理購物相關(guān)問題（商品、訂單、退款等）。請(qǐng)換個(gè)問題試試？"
            print(f"\n??? [護(hù)欄觸發(fā)]: {message}")
            input_items.append({"role": "assistant", "content": message})


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

============================================================
   歡迎使用智能購物助手!
   輸入 'quit' / 'exit' / 'bye' 退出
============================================================

?? 你: 你好，我想買個(gè)手機(jī)

?? [Shopping Triage]: 你好！我是智能購物助手，馬上為你找商品專家！
  ?? [交接] Shopping Triage → Product Specialist

?? [Product Specialist]: 你好！我是商品專家小商 ?? 來幫你挑手機(jī)！
  ?? [Product Specialist]: 正在調(diào)用工具...
  ?? [工具結(jié)果]: 找到以下商品: iPhone 16 Pro (￥8,999), Samsung S26 (￥6,999), Xiaomi 16 (￥3,499)...

?? [Product Specialist]: 我為你找到了幾款熱門手機(jī)：
- iPhone 16 Pro - ￥8,999
- Samsung S26 - ￥6,999
- Xiaomi 16 - ￥3,499

你預(yù)算大概多少？我可以進(jìn)一步推薦！

?? 你: iPhone 16 Pro 怎么樣？

  ?? [Product Specialist]: 正在調(diào)用工具...
  ?? [工具結(jié)果]: A20芯片 | 6.3英寸 | 4800萬像素 | 256GB起 | 電池續(xù)航30小時(shí)

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

要加入購物車嗎？

?? 你: 好的，加一輛

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

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

  ?? [交接] Product Specialist → Order Specialist
  ?? [Order Specialist]: 正在調(diào)用工具...
  ?? [工具結(jié)果]: 訂單 20260315001 狀態(tài): 運(yùn)輸中。預(yù)計(jì)到達(dá): 明天下午。

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

?? 你: 天氣怎么樣？

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

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

from agents import Agent, Runner, trace

# 專家Agent
researcher = Agent(
    name="Researcher",
    instructions="你是一個(gè)調(diào)研專家。對(duì)給定主題進(jìn)行深度調(diào)研，輸出結(jié)構(gòu)化的調(diào)研報(bào)告。",
)

writer = Agent(
    name="Writer",
    instructions="你是一個(gè)資深技術(shù)寫手?；谡{(diào)研報(bào)告撰寫高質(zhì)量的技術(shù)文章。",
)

reviewer = Agent(
    name="Reviewer",
    instructions="""你是一個(gè)嚴(yán)格的審核編輯。從以下維度審核文章:
    1. 技術(shù)準(zhǔn)確性 (0-10)
    2. 可讀性 (0-10)
    3. 完整性 (0-10)
    輸出JSON格式的評(píng)分和修改建議。""",
    output_type=dict,  # 結(jié)構(gòu)化輸出
)

# Manager Agent —— 把專家當(dāng)作工具使用
manager = Agent(
    name="Content Manager",
    instructions="""你是一個(gè)內(nèi)容管理Agent，負(fù)責(zé)協(xié)調(diào)一篇文章的創(chuàng)作。
    工作流程:
    1. 先用 research 工具調(diào)研主題
    2. 再用 write 工具基于調(diào)研結(jié)果撰寫文章
    3. 最后用 review 工具審核文章質(zhì)量
    4. 如果評(píng)分低于8分，根據(jù)意見修改后重新審核
    """,
    tools=[
        researcher.as_tool(
            tool_name="research",
            tool_description="對(duì)指定主題進(jìn)行深度調(diào)研，返回調(diào)研報(bào)告"
        ),
        writer.as_tool(
            tool_name="write",
            tool_description="基于調(diào)研報(bào)告撰寫技術(shù)文章"
        ),
        reviewer.as_tool(
            tool_name="review",
            tool_description="審核文章質(zhì)量，返回評(píng)分和修改建議"
        ),
    ],
)

async def create_article(topic: str) -> str:
    with trace("Content Creation Pipeline"):
        result = await Runner.run(
            manager,
            f"請(qǐng)創(chuàng)作一篇關(guān)于 '{topic}' 的技術(shù)文章"
        )
    return result.final_output

from agents import Agent, Runner, trace

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

# 方式2: 手動(dòng)創(chuàng)建 trace（跨多個(gè) run）
with trace("復(fù)雜工作流", 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: 關(guān)閉 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等場(chǎng)景中刷新trace
        flush_traces()