OpenAI Agents SDK

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="You are a helpful assistant")

result = Runner.run_sync(agent, "写一首关于递归的俳句。")
print(result.final_output)

import asyncio

from pydantic import BaseModel

from agents import Agent, Runner
from agents.decorators import tool


@tool
def get_weather(city: str) -> str:
    """查询城市今天的天气。

    Args:
        city: 城市名，例如 北京。
    """
    return {"北京": "晴，25°C", "上海": "小雨，22°C"}.get(city, "暂无数据")


class WeatherReport(BaseModel):
    city: str
    summary: str
    suggestion: str


agent = Agent(
    name="天气助手",
    instructions="先调用工具查询天气，再给出是否适合户外运动的建议。",
    tools=[get_weather],
    output_type=WeatherReport,  # 让模型输出结构化结果（底层使用 Structured Outputs）
)


async def main() -> None:
    result = await Runner.run(agent, "北京今天适合跑步吗？")
    report = result.final_output_as(WeatherReport)
    print(report.summary, "|", report.suggestion)
    # new_items 记录了本次运行产生的全部条目：工具调用、工具结果、消息……
    print([item.type for item in result.new_items])


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

import asyncio
from dataclasses import dataclass

from agents import Agent, RunContextWrapper, Runner
from agents.decorators import tool


@dataclass
class UserContext:
    name: str
    is_vip: bool


def dynamic_instructions(ctx: RunContextWrapper[UserContext], agent: Agent[UserContext]) -> str:
    level = "VIP" if ctx.context.is_vip else "普通"
    return f"用户名是 {ctx.context.name}，是{level}用户。请礼貌地回答。"


@tool
def get_points(ctx: RunContextWrapper[UserContext]) -> str:
    """查询当前用户的积分。"""
    return f"{ctx.context.name} 当前有 1200 积分"  # 真实场景下这里会查数据库


agent = Agent[UserContext](name="客服", instructions=dynamic_instructions, tools=[get_points])


async def main() -> None:
    result = await Runner.run(agent, "我有多少积分？", context=UserContext(name="小王", is_vip=True))
    print(result.final_output)


asyncio.run(main())

import asyncio
from pathlib import Path

from agents import Agent, Runner
from agents.mcp import MCPServerStdio


async def main() -> None:
    docs_dir = Path(__file__).parent / "docs"
    async with MCPServerStdio(
        name="Filesystem Server via npx",
        params={
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", str(docs_dir)],
        },
    ) as server:  # 上下文管理器负责启动、关闭子进程
        agent = Agent(
            name="文档助手",
            instructions="使用 MCP 提供的文件工具回答问题。",
            mcp_servers=[server],
        )
        result = await Runner.run(agent, "列出你能访问的文件。")
        print(result.final_output)


asyncio.run(main())

import asyncio

from agents import Agent, Runner

history_tutor = Agent(
    name="History Tutor",
    handoff_description="负责历史问题的专家",
    instructions="清晰、简洁地回答历史问题。",
)
math_tutor = Agent(
    name="Math Tutor",
    handoff_description="负责数学问题的专家",
    instructions="一步一步讲解数学题，并给出例题。",
)
triage = Agent(
    name="Triage Agent",
    instructions="把每个作业问题路由给合适的专家。",
    handoffs=[history_tutor, math_tutor],
)


async def main() -> None:
    result = await Runner.run(triage, "美国第一任总统是谁？")
    print(result.final_output)
    print("回答者：", result.last_agent.name)  # 实际给出回答的 agent


asyncio.run(main())

import asyncio

from agents import Agent, Runner

spanish = Agent(name="Spanish agent", instructions="把用户的消息翻译成西班牙语")
french = Agent(name="French agent", instructions="把用户的消息翻译成法语")

orchestrator = Agent(
    name="orchestrator",
    instructions="你是翻译助手，使用给你的工具完成翻译；需要多种语言时就调用多个工具。",
    tools=[
        spanish.as_tool(tool_name="translate_to_spanish", tool_description="翻译成西班牙语"),
        french.as_tool(tool_name="translate_to_french", tool_description="翻译成法语"),
    ],
)


async def main() -> None:
    result = await Runner.run(orchestrator, "把 'Hello, how are you?' 翻译成西班牙语和法语。")
    print(result.final_output)


asyncio.run(main())

import asyncio

from pydantic import BaseModel

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
)
from agents.decorators import input_guardrail


class HomeworkCheck(BaseModel):
    is_math_homework: bool
    reasoning: str


# 用一个便宜、快速的 agent 做检查
guardrail_agent = Agent(
    name="Guardrail check",
    instructions="判断用户是不是在让你代写数学作业。",
    output_type=HomeworkCheck,
)


@input_guardrail
async def math_guardrail(
    ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
    result = await Runner.run(guardrail_agent, input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=result.final_output_as(HomeworkCheck).is_math_homework,
    )


support_agent = Agent(
    name="Customer support agent",
    instructions="你是客服，帮助客户解决问题。",
    input_guardrails=[math_guardrail],
)


async def main() -> None:
    try:
        await Runner.run(support_agent, "帮我解一下 2x + 3 = 11")
    except InputGuardrailTripwireTriggered:
        print("触发了“代写作业”护栏，请求被拦截")


asyncio.run(main())

import asyncio

from agents import Agent, Runner, SQLiteSession


async def main() -> None:
    agent = Agent(name="Assistant", instructions="回答要非常简洁。")
    session = SQLiteSession("conversation_123")  # 默认是内存数据库；也可以传入文件路径持久化

    r1 = await Runner.run(agent, "金门大桥在哪个城市？", session=session)
    print(r1.final_output)  # 旧金山
    r2 = await Runner.run(agent, "它在哪个州？", session=session)  # 自动带上之前的历史
    print(r2.final_output)  # 加利福尼亚


asyncio.run(main())

import asyncio

from agents import Agent, ItemHelpers, Runner
from agents.decorators import tool


@tool
def how_many_jokes() -> int:
    """返回要讲的笑话数量。"""
    return 3


async def main() -> None:
    agent = Agent(
        name="Joker",
        instructions="先调用 how_many_jokes，再讲对应数量的笑话。",
        tools=[how_many_jokes],
    )
    result = Runner.run_streamed(agent, input="讲几个笑话")
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            continue  # 原始的 token 级事件，这里忽略
        elif event.type == "agent_updated_stream_event":
            print(f"当前 agent：{event.new_agent.name}")
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("-- 调用了工具")
            elif event.item.type == "tool_call_output_item":
                print(f"-- 工具输出：{event.item.output}")
            elif event.item.type == "message_output_item":
                print(ItemHelpers.text_message_output(event.item))


asyncio.run(main())

import asyncio

from agents import Agent, Runner
from agents.decorators import tool


@tool(needs_approval=True)  # 也可以传一个异步函数，按参数逐次判断是否需要审批
async def cancel_order(order_id: int) -> str:
    """取消订单。"""
    return f"订单 {order_id} 已取消"


agent = Agent(name="Support agent", instructions="处理工单，必要时取消订单。", tools=[cancel_order])


async def main() -> None:
    result = await Runner.run(agent, "请取消我的订单 42")
    while result.interruptions:  # 有待审批的工具调用
        state = result.to_state()  # RunState 可以序列化，跨进程、跨天后再恢复都可以
        for item in result.interruptions:
            ok = input(f"批准 {item.name}({item.arguments})？[y/N] ") == "y"
            if ok:
                state.approve(item)
            else:
                state.reject(item)
        result = await Runner.run(agent, state)  # 从暂停处继续
    print(result.final_output)


asyncio.run(main())

from agents import Agent, Runner, trace

agent = Agent(name="Assistant", instructions="简洁回答。")

with trace(workflow_name="多轮对话示例", group_id="thread_123"):  # 把多次运行归到同一个 trace 下
    first = Runner.run_sync(agent, "金门大桥在哪个城市？")
    second = Runner.run_sync(
        agent, first.to_input_list() + [{"role": "user", "content": "在哪个州？"}]
    )
    print(second.final_output)