flowchart TB
    subgraph "Runner 模块"
        RUNNER[Runner 静态类]
        AGENTRUNNER[AgentRunner 实例]
        
        subgraph "核心方法"
            RUN[run 异步执行]
            RUNSYNC[run_sync 同步执行]
            RUNSTREAMED[run_streamed 流式执行]
        end
        
        subgraph "配置与数据"
            RUNCONFIG[RunConfig 运行配置]
            RUNOPTIONS[RunOptions 执行选项]
            MODELINPUTDATA[ModelInputData 模型输入]
            CALLMODELDATA[CallModelData 模型调用数据]
        end
        
        subgraph "状态追踪"
            SERVERTRACKER[_ServerConversationTracker]
            TOOLTRACXER[AgentToolUseTracker]
        end
    end
    
    subgraph "执行引擎"
        RUNIMPL[RunImpl 核心引擎]
        TRACEMANAGER[TraceCtxManager 追踪管理]
        SINGLESTEP[SingleStepResult 单步结果]
        NEXTSTEP[NextStep* 执行步骤]
    end
    
    subgraph "外部依赖"
        AGENT[Agent 代理]
        SESSION[Session 会话]
        MODEL[Model 模型]
        TOOLS[Tools 工具]
        GUARDRAILS[Guardrails 安全防护]
        HOOKS[Hooks 生命周期]
        TRACING[Tracing 追踪]
    end
    
    RUNNER --> RUN
    RUNNER --> RUNSYNC
    RUNNER --> RUNSTREAMED
    
    RUN --> AGENTRUNNER
    RUNSYNC --> AGENTRUNNER
    RUNSTREAMED --> AGENTRUNNER
    
    AGENTRUNNER --> RUNCONFIG
    AGENTRUNNER --> RUNOPTIONS
    AGENTRUNNER --> SERVERTRACKER
    AGENTRUNNER --> TOOLTRACXER
    
    AGENTRUNNER --> RUNIMPL
    RUNIMPL --> TRACEMANAGER
    RUNIMPL --> SINGLESTEP
    RUNIMPL --> NEXTSTEP
    
    RUNIMPL --> AGENT
    RUNIMPL --> SESSION
    RUNIMPL --> MODEL
    RUNIMPL --> TOOLS
    RUNIMPL --> GUARDRAILS
    RUNIMPL --> HOOKS
    RUNIMPL --> TRACING
    
    RUNCONFIG --> MODELINPUTDATA
    RUNCONFIG --> CALLMODELDATA
    
    style RUNNER fill:#e1f5fe
    style RUNIMPL fill:#f3e5f5
    style AGENTRUNNER fill:#e8f5e8

async def run(self, starting_agent: Agent[TContext], input: str | list[TResponseInputItem], **kwargs) -> RunResult:
    """主执行循环的简化逻辑"""
    # 1) 初始化执行环境
    context = kwargs.get("context")
    max_turns = kwargs.get("max_turns", DEFAULT_MAX_TURNS)
    run_config = kwargs.get("run_config") or RunConfig()
    session = kwargs.get("session")
    
    # 2) 准备输入数据
    prepared_input = await self._prepare_input_with_session(input, session, run_config.session_input_callback)
    
    # 3) 创建执行追踪器
    tool_use_tracker = AgentToolUseTracker()
    server_conversation_tracker = _ServerConversationTracker(...)
    
    # 4) 启动追踪上下文
    with TraceCtxManager(workflow_name=run_config.workflow_name) as trace_ctx:
        # 5) 创建执行引擎实例
        run_impl = RunImpl(
            run_config=run_config,
            hooks=hooks,
            tool_use_tracker=tool_use_tracker,
            server_conversation_tracker=server_conversation_tracker,
            trace_ctx=trace_ctx
        )
        
        # 6) 执行主循环
        return await run_impl.run(
            starting_agent=starting_agent,
            original_input=input,
            prepared_input=prepared_input,
            context=context,
            max_turns=max_turns
        )

async def _prepare_input_with_session(
    self,
    input: str | list[TResponseInputItem],
    session: Session | None,
    session_input_callback: SessionInputCallback | None
) -> list[TResponseInputItem]:
    """集成会话历史的输入准备算法"""
    
    # 1) 转换用户输入为标准格式
    user_input_items = ItemHelpers.input_to_new_input_list(input)
    
    if session is None:
        return user_input_items
    
    # 2) 从会话加载历史消息
    try:
        session_items = await session.get_messages()
    except Exception as e:
        logger.warning(f"Failed to load session messages: {e}")
        session_items = []
    
    # 3) 应用会话输入回调（如果配置）
    if session_input_callback:
        try:
            if inspect.iscoroutinefunction(session_input_callback):
                session_items = await session_input_callback(session_items, user_input_items)
            else:
                session_items = session_input_callback(session_items, user_input_items)
        except Exception as e:
            logger.error(f"Session input callback failed: {e}")
    
    # 4) 合并历史消息和当前输入
    combined_input = []
    combined_input.extend(session_items)
    combined_input.extend(user_input_items)
    
    return combined_input

class _ServerConversationTracker:
    """OpenAI 服务器对话状态同步算法"""
    
    def prepare_input(
        self,
        original_input: str | list[TResponseInputItem],
        generated_items: list[RunItem],
    ) -> list[TResponseInputItem]:
        """准备发送到服务器的输入，避免重复发送"""
        input_items: list[TResponseInputItem] = []
        
        # 首次调用时包含原始输入
        if not generated_items:
            input_items.extend(ItemHelpers.input_to_new_input_list(original_input))
        
        # 处理生成的项目，跳过已发送或来自服务器的项目
        for item in generated_items:
            raw_item_id = id(item.raw_item)
            
            # 跳过已发送或服务器生成的项目
            if raw_item_id in self.sent_items or raw_item_id in self.server_items:
                continue
                
            input_items.append(item.to_input_item())
            self.sent_items.add(raw_item_id)  # 标记为已发送
        
        return input_items
    
    def track_server_items(self, model_response: ModelResponse) -> None:
        """追踪服务器返回的项目，避免重复发送"""
        for output_item in model_response.output:
            self.server_items.add(id(output_item))
        
        # 更新 previous_response_id（仅在使用该模式时）
        if (self.conversation_id is None
            and self.previous_response_id is not None
            and model_response.response_id is not None):
            self.previous_response_id = model_response.response_id

@dataclass
class RunConfig:
    """全局运行配置，影响整个执行过程"""
    
    # 模型配置
    model: str | Model | None = None                    # 覆盖代理模型设置
    model_provider: ModelProvider = field(default_factory=MultiProvider)  # 模型提供商
    model_settings: ModelSettings | None = None        # 全局模型参数
    
    # 输入输出过滤
    call_model_input_filter: CallModelInputFilter | None = None   # 模型输入预处理
    call_model_output_filter: CallModelOutputFilter | None = None # 模型输出后处理
    
    # 安全防护
    input_guardrails: list[InputGuardrail[Any]] | None = None     # 全局输入安全检查
    output_guardrails: list[OutputGuardrail[Any]] | None = None   # 全局输出安全检查
    
    # 会话管理
    session_input_callback: SessionInputCallback | None = None    # 会话输入处理回调
    
    # 切换配置
    handoff_input_filter: HandoffInputFilter | None = None       # 全局切换输入过滤
    
    # 追踪配置
    workflow_name: str | None = None                             # 工作流名称
    disable_tracing: bool = False                                # 禁用追踪
    trace_include_sensitive_data: bool = True                    # 包含敏感数据

# 生产环境配置
production_config = RunConfig(
    model_settings=ModelSettings(temperature=0.3),  # 降低随机性
    disable_tracing=False,                          # 启用追踪
    trace_include_sensitive_data=False,             # 不包含敏感数据
    input_guardrails=[content_safety_guardrail],    # 启用安全检查
    output_guardrails=[output_safety_guardrail]
)

# 开发调试配置
debug_config = RunConfig(
    disable_tracing=False,                          # 启用详细追踪
    trace_include_sensitive_data=True,              # 包含完整数据
    workflow_name="debug_session"                   # 便于追踪识别
)

# 高性能配置
performance_config = RunConfig(
    disable_tracing=True,                           # 禁用追踪减少开销
    model_settings=ModelSettings(max_tokens=1000), # 限制输出长度
    call_model_input_filter=input_compression_filter # 压缩输入
)

import asyncio
from agents import Agent, Runner

async def basic_usage():
    """基础使用：简单对话"""
    agent = Agent(
        name="Assistant",
        instructions="你是一个有用的助手"
    )
    
    result = await Runner.run(agent, "你好！")
    print(result.final_output)
    
    # 查看执行统计
    print(f"使用令牌: {result.usage.total_tokens}")
    print(f"执行轮次: {len(result.all_items)}")

asyncio.run(basic_usage())

from agents import Agent, Runner, SQLiteSession

async def session_management():
    """会话管理：维护对话历史"""
    agent = Agent(
        name="ChatBot",
        instructions="维持连贯的对话，记住之前的交流内容"
    )
    
    # 创建持久化会话
    session = SQLiteSession("chat_history.db")
    
    # 多轮对话
    questions = [
        "我叫张三，今年30岁",
        "我的年龄是多少？",
        "我的名字是什么？"
    ]
    
    for question in questions:
        result = await Runner.run(
            agent,
            question,
            session=session  # 自动加载和保存历史
        )
        print(f"Q: {question}")
        print(f"A: {result.final_output}\n")
    
    await session.close()

asyncio.run(session_management())

from agents import Agent, Runner

def streaming_execution():
    """流式执行：实时获取执行事件"""
    agent = Agent(
        name="StreamingAgent",
        instructions="详细解释你的思考过程"
    )
    
    # 启动流式执行
    result = Runner.run_streamed(
        agent,
        "解释量子计算的基本原理"
    )
    
    print("开始流式执行...")
    
    # 实时处理事件
    for event in result.stream_events():
        if event.type == "agent_updated":
            print(f"代理切换: {event.data.agent.name}")
        elif event.type == "run_item":
            if event.data.item.type == "message_output":
                print(f"输出: {event.data.item.content}")
        elif event.type == "raw_responses":
            print(f"模型响应: {len(event.data.responses)} 个响应")
    
    print(f"最终结果: {result.final_output}")

streaming_execution()

from agents import Agent, Runner, RunConfig
from agents.exceptions import MaxTurnsExceeded, InputGuardrailTripwireTriggered

async def error_handling():
    """错误处理：优雅处理各种异常情况"""
    agent = Agent(
        name="ReliableAgent",
        instructions="尽力完成任务，但要注意安全边界"
    )
    
    config = RunConfig(
        max_turns=5,  # 限制最大轮次
        disable_tracing=False  # 启用追踪便于调试
    )
    
    try:
        result = await Runner.run(
            agent,
            "帮我写一个复杂的程序",
            run_config=config,
            max_turns=3
        )
        return result
        
    except MaxTurnsExceeded as e:
        print(f"达到最大轮次限制: {e.details.max_turns}")
        print(f"当前执行历史: {len(e.details.items)} 项")
        # 可以返回部分结果
        return e.details.partial_result
        
    except InputGuardrailTripwireTriggered as e:
        print(f"输入安全检查失败: {e.guardrail_name}")
        print(f"失败原因: {e.failure_reason}")
        # 记录安全事件
        return None
        
    except Exception as e:
        print(f"未预期错误: {e}")
        # 记录错误日志
        return None

result = await error_handling()

from agents import Agent, Runner, RunConfig, RunHooks

class CustomRunHooks(RunHooks):
    """自定义执行钩子"""
    
    async def on_agent_start(self, context, agent):
        print(f"开始执行代理: {agent.name}")
    
    async def on_model_call_start(self, context, agent, input_data):
        print(f"模型调用开始，输入长度: {len(input_data.input)}")
    
    async def on_tool_execution_start(self, context, agent, tool_name, tool_input):
        print(f"工具执行开始: {tool_name}")
    
    async def on_agent_end(self, context, agent, result):
        print(f"代理执行完成: {agent.name}, 令牌使用: {result.usage.total_tokens}")

async def advanced_customization():
    """高级定制：完全控制执行流程"""
    
    def custom_input_filter(call_data):
        """自定义输入过滤"""
        # 添加系统提示
        system_prompt = "请在回答前先思考问题的关键点。"
        call_data.model_data.instructions = system_prompt
        return call_data.model_data
    
    def custom_output_filter(response):
        """自定义输出过滤"""
        # 添加输出格式化
        if hasattr(response, 'final_output'):
            response.final_output = f"[AI回答] {response.final_output}"
        return response
    
    agent = Agent(
        name="CustomAgent",
        instructions="你是一个定制化的AI助手"
    )
    
    config = RunConfig(
        call_model_input_filter=custom_input_filter,
        call_model_output_filter=custom_output_filter,
        workflow_name="custom_workflow"
    )
    
    result = await Runner.run(
        agent,
        "解释机器学习的基本概念",
        hooks=CustomRunHooks(),
        run_config=config
    )
    
    return result

result = await advanced_customization()

Runner (执行调度器)
    ├── run() - 标准异步执行
    ├── run_streamed() - 流式异步执行
    └── run_sync() - 同步阻塞执行

RunConfig (执行配置)
    ├── 模型配置
    ├── 安全防护配置
    ├── 生命周期钩子
    └── 服务器对话管理

RunResult (执行结果)
    ├── final_output - 最终输出
    ├── new_items - 生成的历史项
    ├── raw_responses - 原始模型响应
    └── guardrail_results - 防护检查结果

RunResultStreaming (流式结果)
    ├── stream_events() - 流式事件生成器
    ├── current_agent - 当前执行代理
    └── is_complete - 完成状态

@staticmethod
async def run(
    agent: Agent[TContext],
    input: str | list[TResponseInputItem],
    session: Session | None = None,
    run_config: RunConfig | None = None,
    context: TContext | None = None,
) -> RunResult

@dataclass
class RunResult:
    input: str | list[TResponseInputItem]  # 原始输入
    new_items: list[RunItem]               # 新生成的运行项
    raw_responses: list[ModelResponse]     # 原始模型响应列表
    final_output: Any                      # 最终输出（通常是字符串）
    last_agent: Agent[Any]                 # 最后执行的代理
    
    # 防护检查结果
    input_guardrail_results: list[InputGuardrailResult]
    output_guardrail_results: list[OutputGuardrailResult]
    tool_input_guardrail_results: list[ToolInputGuardrailResult]
    tool_output_guardrail_results: list[ToolOutputGuardrailResult]
    
    # 上下文包装器
    context_wrapper: RunContextWrapper[Any]
    
    # 便捷方法
    def final_output_as(cls: type[T]) -> T
    def to_input_list() -> list[TResponseInputItem]
    
    # 属性
    @property
    def last_response_id() -> str | None

from agents import Agent, Runner
from agents.memory import SQLiteSession

async def basic_run_example():
    """基础执行示例"""
    
    # 创建代理
    agent = Agent(
        name="Assistant",
        instructions="你是一个友好的AI助手。"
    )
    
    # 执行代理
    result = await Runner.run(
        agent=agent,
        input="你好，请介绍一下自己。"
    )
    
    # 访问结果
    print(f"最终输出: {result.final_output}")
    print(f"生成了 {len(result.new_items)} 个新项目")
    print(f"调用模型 {len(result.raw_responses)} 次")
    print(f"最后代理: {result.last_agent.name}")

async def run_with_session_example():
    """带会话历史的执行示例"""
    
    agent = Agent(
        name="ChatBot",
        instructions="你是一个记忆良好的聊天机器人。"
    )
    
    # 创建会话
    session = SQLiteSession("user_123", db_path="chat.db")
    
    # 第一轮对话
    result1 = await Runner.run(
        agent=agent,
        input="我叫张三，我喜欢编程。",
        session=session
    )
    print(f"助手: {result1.final_output}")
    
    # 第二轮对话（有历史上下文）
    result2 = await Runner.run(
        agent=agent,
        input="你还记得我的名字吗？",
        session=session
    )
    print(f"助手: {result2.final_output}")
    # 预期输出: "当然记得，你叫张三..."

async def run_with_tools_example():
    """带工具的执行示例"""
    
    from agents import function_tool
    
    @function_tool
    def calculate(expression: str) -> str:
        """计算数学表达式"""
        return str(eval(expression))
    
    agent = Agent(
        name="MathBot",
        instructions="你是一个数学助手，使用calculate工具进行计算。",
        tools=[calculate]
    )
    
    result = await Runner.run(
        agent=agent,
        input="请计算 123 * 456 的结果"
    )
    
    print(f"最终输出: {result.final_output}")
    
    # 查看生成的项目
    for item in result.new_items:
        if hasattr(item, 'tool_name'):
            print(f"调用工具: {item.tool_name}")
        elif hasattr(item, 'content'):
            print(f"消息: {item.content}")

async def run_with_config_example():
    """带配置的执行示例"""
    
    from agents import RunConfig
    
    agent = Agent(
        name="ConfiguredAgent",
        instructions="遵循配置运行的代理。"
    )
    
    # 创建配置
    config = RunConfig(
        model="gpt-4o",  # 指定模型
        max_turns=5,     # 最多5轮对话
        trace_include_sensitive_data=False  # 不包含敏感数据
    )
    
    result = await Runner.run(
        agent=agent,
        input="请帮我完成一个复杂任务。",
        run_config=config
    )
    
    print(f"使用的模型: {config.model}")
    print(f"实际轮次: {len(result.raw_responses)}")

async def run_with_context_example():
    """带自定义上下文的执行示例"""
    
    from dataclasses import dataclass
    
    @dataclass
    class UserContext:
        user_id: str
        database_connection: Any
        preferences: dict
    
    @function_tool
    def get_user_preference(key: str, context: UserContext) -> str:
        """获取用户偏好设置"""
        return context.preferences.get(key, "未设置")
    
    agent = Agent(
        name="PersonalAssistant",
        instructions="根据用户偏好提供个性化服务。",
        tools=[get_user_preference]
    )
    
    # 创建用户上下文
    user_context = UserContext(
        user_id="user_123",
        database_connection=None,  # 实际应用中是数据库连接
        preferences={"language": "中文", "theme": "dark"}
    )
    
    result = await Runner.run(
        agent=agent,
        input="我的语言偏好是什么？",
        context=user_context
    )
    
    print(f"输出: {result.final_output}")
    print(f"Token使用: {result.context_wrapper.usage}")

@staticmethod
async def run_streamed(
    agent: Agent[TContext],
    input: str | list[TResponseInputItem],
    session: Session | None = None,
    run_config: RunConfig | None = None,
    context: TContext | None = None,
) -> RunResultStreaming

@dataclass
class RunResultStreaming(RunResultBase):
    current_agent: Agent[Any]      # 当前执行的代理
    current_turn: int              # 当前执行轮次
    max_turns: int                 # 最大允许轮次
    final_output: Any              # 最终输出（完成前为None）
    is_complete: bool              # 是否完成执行
    
    # 继承自 RunResultBase
    input: str | list[TResponseInputItem]
    new_items: list[RunItem]
    raw_responses: list[ModelResponse]
    input_guardrail_results: list[InputGuardrailResult]
    output_guardrail_results: list[OutputGuardrailResult]
    tool_input_guardrail_results: list[ToolInputGuardrailResult]
    tool_output_guardrail_results: list[ToolOutputGuardrailResult]
    context_wrapper: RunContextWrapper[Any]
    
    # 流式方法
    async def stream_events() -> AsyncIterator[StreamEvent]
    def cancel() -> None

# StreamEvent 是联合类型
StreamEvent = (
    RawResponsesStreamEvent |      # 原始模型响应事件
    RunItemStreamEvent |           # 运行项事件
    AgentUpdatedStreamEvent        # 代理更新事件
)

# RunItemStreamEvent 的事件名称
event_names = [
    "message_output_created",      # 消息输出创建
    "tool_called",                 # 工具被调用
    "tool_output",                 # 工具输出
    "handoff_requested",           # 代理切换请求
    "handoff_occured",             # 代理切换发生
    "reasoning_item_created",      # 推理项创建
    "mcp_approval_requested",      # MCP批准请求
    "mcp_list_tools"               # MCP工具列表
]

async def streamed_basic_example():
    """基础流式执行示例"""
    
    agent = Agent(
        name="StreamingAssistant",
        instructions="你是一个流式响应助手。"
    )
    
    # 启动流式执行
    result = await Runner.run_streamed(
        agent=agent,
        input="请详细介绍一下人工智能的发展历史。"
    )
    
    # 处理流式事件
    accumulated_text = ""
    
    async for event in result.stream_events():
        if event.type == "run_item_stream_event":
            if event.name == "message_output_created":
                # 消息输出事件
                content = event.item.content
                
                # 计算增量内容
                if content != accumulated_text:
                    delta = content[len(accumulated_text):]
                    print(delta, end="", flush=True)
                    accumulated_text = content
            
            elif event.name == "tool_called":
                # 工具调用事件
                print(f"\n[调用工具: {event.item.tool_name}]")
            
            elif event.name == "tool_output":
                # 工具输出事件
                print(f"[工具结果: {event.item.output[:50]}...]")
        
        elif event.type == "agent_updated_stream_event":
            # 代理更新事件
            print(f"\n[切换到代理: {event.new_agent.name}]")
    
    print(f"\n\n最终输出: {result.final_output}")

async def streamed_with_ui_example():
    """流式执行与UI集成示例"""
    
    from typing import Callable
    
    class ChatUI:
        """模拟的聊天UI类"""
        
        def __init__(self):
            self.messages = []
            self.current_message = ""
        
        def append_to_current_message(self, text: str):
            """追加文本到当前消息"""
            self.current_message += text
            # 实际应用中这里会更新UI
            print(text, end="", flush=True)
        
        def finish_current_message(self):
            """完成当前消息"""
            self.messages.append(self.current_message)
            self.current_message = ""
            print()  # 换行
        
        def show_tool_call(self, tool_name: str, args: dict):
            """显示工具调用"""
            print(f"\n🔧 正在使用工具: {tool_name}")
            print(f"   参数: {args}")
        
        def show_tool_result(self, result: str):
            """显示工具结果"""
            print(f"✅ 工具结果: {result[:100]}...")
    
    async def run_with_ui(agent: Agent, user_input: str):
        """带UI的流式执行"""
        
        ui = ChatUI()
        
        result = await Runner.run_streamed(
            agent=agent,
            input=user_input
        )
        
        accumulated_content = ""
        
        async for event in result.stream_events():
            if event.type == "run_item_stream_event":
                if event.name == "message_output_created":
                    # 增量文本输出
                    content = event.item.content
                    if content != accumulated_content:
                        delta = content[len(accumulated_content):]
                        ui.append_to_current_message(delta)
                        accumulated_content = content
                
                elif event.name == "tool_called":
                    # 显示工具调用
                    ui.show_tool_call(
                        event.item.tool_name,
                        event.item.arguments
                    )
                
                elif event.name == "tool_output":
                    # 显示工具结果
                    ui.show_tool_result(event.item.output)
        
        ui.finish_current_message()
        return result
    
    # 使用示例
    agent = Agent(
        name="UIAssistant",
        instructions="你是一个用户界面助手。"
    )
    
    result = await run_with_ui(agent, "请帮我查询今天的天气")

async def streamed_with_cancellation_example():
    """带取消功能的流式执行示例"""
    
    import asyncio
    
    agent = Agent(
        name="LongRunningAgent",
        instructions="你会进行长时间的处理。"
    )
    
    result = await Runner.run_streamed(
        agent=agent,
        input="请进行一个非常详细的分析。"
    )
    
    # 设置超时取消
    async def cancel_after_timeout(seconds: float):
        """N秒后取消执行"""
        await asyncio.sleep(seconds)
        if not result.is_complete:
            print(f"\n[超时 {seconds}秒，取消执行]")
            result.cancel()
    
    # 启动超时任务
    timeout_task = asyncio.create_task(cancel_after_timeout(5.0))
    
    try:
        async for event in result.stream_events():
            # 处理事件
            if event.type == "run_item_stream_event":
                if event.name == "message_output_created":
                    print(".", end="", flush=True)
    except asyncio.CancelledError:
        print("\n执行已取消")
    finally:
        timeout_task.cancel()

async def streamed_error_handling_example():
    """流式执行的错误处理示例"""
    
    agent = Agent(
        name="ErrorProneAgent",
        instructions="可能会遇到错误的代理。"
    )
    
    result = await Runner.run_streamed(
        agent=agent,
        input="执行可能失败的任务"
    )
    
    try:
        async for event in result.stream_events():
            # 处理事件
            if event.type == "run_item_stream_event":
                print(f"事件: {event.name}")
    
    except MaxTurnsExceeded as e:
        print(f"超过最大轮次: {e}")
        print(f"已生成 {len(e.run_data.new_items)} 个项目")
    
    except InputGuardrailTripwireTriggered as e:
        print(f"输入防护触发: {e.guardrail_result.output.message}")
    
    except OutputGuardrailTripwireTriggered as e:
        print(f"输出防护触发: {e.guardrail_result.output.message}")
    
    except Exception as e:
        print(f"执行错误: {e}")

@staticmethod
def run_sync(
    agent: Agent[TContext],
    input: str | list[TResponseInputItem],
    session: Session | None = None,
    run_config: RunConfig | None = None,
    context: TContext | None = None,
) -> RunResult

def sync_basic_example():
    """同步执行基础示例"""
    
    from agents import Agent, Runner
    
    agent = Agent(
        name="SyncAssistant",
        instructions="同步执行的助手。"
    )
    
    # 同步执行（阻塞）
    result = Runner.run_sync(
        agent=agent,
        input="你好，世界！"
    )
    
    print(f"输出: {result.final_output}")

def sync_in_script_example():
    """在脚本中使用同步执行"""
    
    # 不需要 async/await 语法
    if __name__ == "__main__":
        agent = Agent(name="ScriptAgent", instructions="脚本助手")
        result = Runner.run_sync(agent, "执行任务")
        print(result.final_output)

def sync_with_traditional_code_example():
    """与传统同步代码集成"""
    
    def legacy_function():
        """传统的同步函数"""
        agent = Agent(name="LegacyAgent", instructions="传统代码助手")
        
        # 可以直接调用，无需异步上下文
        result = Runner.run_sync(agent, "处理请求")
        
        return result.final_output
    
    # 调用
    output = legacy_function()
    print(output)

@dataclass
class RunConfig:
    def __init__(
        self,
        model: str | Model | None = None,
        model_provider: ModelProvider = MultiProvider(),
        model_settings: ModelSettings | None = None,
        handoff_input_filter: HandoffInputFilter | None = None,
        input_guardrails: list[InputGuardrail[Any]] | None = None,
        output_guardrails: list[OutputGuardrail[Any]] | None = None,
        max_turns: int = DEFAULT_MAX_TURNS,  # 默认10
        trace_include_sensitive_data: bool = True,
        call_model_input_filter: CallModelInputFilter | None = None,
        session_input_callback: SessionInputCallback | None = None,
        hooks: RunHooksBase | None = None,
        conversation_id: str | None = None,
        previous_response_id: str | None = None,
    )

from agents import RunConfig, ModelSettings
from agents.guardrail import InputGuardrail, OutputGuardrail

# 基础配置
basic_config = RunConfig(
    model="gpt-4o",
    max_turns=15
)

# 完整配置
advanced_config = RunConfig(
    # 模型配置
    model="gpt-4o",
    model_settings=ModelSettings(
        temperature=0.7,
        max_tokens=2000,
        top_p=0.9
    ),
    
    # 安全防护
    input_guardrails=[ContentModerationGuardrail()],
    output_guardrails=[SensitiveInfoGuardrail()],
    
    # 执行控制
    max_turns=20,
    trace_include_sensitive_data=False,
    
    # 生命周期钩子
    hooks=MyCustomHooks()
)

# 服务器对话配置
server_conversation_config = RunConfig(
    conversation_id="conv_abc123",  # 使用现有对话
    model="gpt-4o"
)

# 字符串模型名称
config = RunConfig(model="gpt-4o")
config = RunConfig(model="gpt-4o-mini")
config = RunConfig(model="gpt-3.5-turbo")

# 自定义模型实例
from agents.models import CustomModel
custom_model = CustomModel(...)
config = RunConfig(model=custom_model)

# 覆盖代理的模型设置
agent = Agent(name="A", model="gpt-3.5-turbo")
config = RunConfig(model="gpt-4o")
result = await Runner.run(agent, "test", run_config=config)
# 实际使用 gpt-4o（配置优先）

from agents import ModelSettings

# 创建性思维配置
creative_settings = ModelSettings(
    temperature=1.2,      # 高随机性
    top_p=0.95,           # 多样性
    max_tokens=3000       # 较长输出
)

# 精确性配置
precise_settings = ModelSettings(
    temperature=0.1,      # 低随机性
    top_p=0.5,            # 集中采样
    max_tokens=1000       # 简洁输出
)

# 应用配置
config = RunConfig(model_settings=creative_settings)
result = await Runner.run(agent, input, run_config=config)

from agents.guardrail import InputGuardrail, OutputGuardrail

class ContentModerationGuardrail(InputGuardrail):
    """内容审核防护"""
    async def run(self, input_text: str, context: Any):
        # 检查不当内容
        if contains_inappropriate_content(input_text):
            return InputGuardrailResult(
                output=GuardrailFunctionOutput(
                    tripwire_triggered=True,
                    message="输入包含不当内容"
                )
            )
        return InputGuardrailResult(
            output=GuardrailFunctionOutput(tripwire_triggered=False)
        )

class PIIDetectionGuardrail(OutputGuardrail):
    """个人信息检测防护"""
    async def run(self, output_text: str, context: Any):
        # 检测个人身份信息
        if contains_pii(output_text):
            return OutputGuardrailResult(
                output=GuardrailFunctionOutput(
                    tripwire_triggered=True,
                    message="输出包含个人隐私信息"
                )
            )
        return OutputGuardrailResult(
            output=GuardrailFunctionOutput(tripwire_triggered=False)
        )

# 配置防护
config = RunConfig(
    input_guardrails=[ContentModerationGuardrail()],
    output_guardrails=[PIIDetectionGuardrail()]
)

# 简单任务：较少轮次
simple_config = RunConfig(max_turns=5)

# 复杂任务：较多轮次
complex_config = RunConfig(max_turns=20)

# 无限制（不推荐）
unlimited_config = RunConfig(max_turns=9999)

# 超过轮次会抛出异常
try:
    result = await Runner.run(agent, input, run_config=simple_config)
except MaxTurnsExceeded as e:
    print(f"超过最大轮次 {e.run_data.new_items}")

from agents.lifecycle import RunHooksBase

class CustomHooks(RunHooksBase):
    """自定义生命周期钩子"""
    
    async def on_run_start(self, agent, input, context):
        """执行开始时调用"""
        print(f"开始执行代理: {agent.name}")
    
    async def on_run_end(self, result, context):
        """执行结束时调用"""
        print(f"执行完成，输出: {result.final_output}")
    
    async def on_tool_call(self, tool_name, arguments, context):
        """工具调用前调用"""
        print(f"调用工具: {tool_name}")
    
    async def on_tool_result(self, tool_name, result, context):
        """工具执行后调用"""
        print(f"工具结果: {result}")
    
    async def on_agent_switch(self, from_agent, to_agent, context):
        """代理切换时调用"""
        print(f"切换代理: {from_agent.name} -> {to_agent.name}")

# 使用钩子
config = RunConfig(hooks=CustomHooks())
result = await Runner.run(agent, input, run_config=config)

result = await Runner.run(agent, input)

# 最终输出
print(result.final_output)  # "这是助手的回复"

# 原始输入
print(result.input)  # "用户的问题"

# 生成的新项目
for item in result.new_items:
    print(type(item).__name__)  # MessageOutputItem, ToolCallItem等

# 原始模型响应
for response in result.raw_responses:
    print(response.response_id)  # "resp_abc123"

# 最后执行的代理
print(result.last_agent.name)  # "FinalAgent"

# 防护检查结果
print(len(result.input_guardrail_results))   # 输入防护数量
print(len(result.output_guardrail_results))  # 输出防护数量
print(len(result.tool_input_guardrail_results))   # 工具输入防护
print(len(result.tool_output_guardrail_results))  # 工具输出防护

# 上下文包装器
print(result.context_wrapper.usage)  # Token使用统计
print(result.context_wrapper.context)  # 用户自定义上下文

from dataclasses import dataclass

@dataclass
class WeatherData:
    temperature: float
    condition: str
    humidity: int

# 配置代理返回结构化输出
agent = Agent(
    name="WeatherAgent",
    output_schema=WeatherData
)

result = await Runner.run(agent, "查询天气")

# 类型安全的转换
weather: WeatherData = result.final_output_as(WeatherData)
print(f"温度: {weather.temperature}°C")
print(f"状况: {weather.condition}")

# 带类型检查的转换
try:
    weather = result.final_output_as(WeatherData, raise_if_incorrect_type=True)
except TypeError as e:
    print(f"类型不匹配: {e}")

# 第一轮对话
result1 = await Runner.run(agent, "第一个问题")

# 将结果转换为新的输入列表
input_list = result1.to_input_list()

# 第二轮对话，使用转换后的输入（包含历史）
result2 = await Runner.run(agent, "第二个问题")

# 等价于使用会话
session = SQLiteSession("user_123")
result1 = await Runner.run(agent, "第一个问题", session=session)
result2 = await Runner.run(agent, "第二个问题", session=session)

result = await Runner.run(agent, input)

response_id = result.last_response_id
if response_id:
    print(f"最后响应ID: {response_id}")
    
    # 可用于服务器端对话状态
    next_config = RunConfig(previous_response_id=response_id)
    next_result = await Runner.run(agent, "下一个问题", run_config=next_config)

async def stream_events(self) -> AsyncIterator[StreamEvent]

result = await Runner.run_streamed(agent, input)

async for event in result.stream_events():
    # 类型检查和处理
    if event.type == "run_item_stream_event":
        # 运行项事件
        if event.name == "message_output_created":
            print(f"消息: {event.item.content}")
        
        elif event.name == "tool_called":
            print(f"工具: {event.item.tool_name}")
        
        elif event.name == "tool_output":
            print(f"结果: {event.item.output}")
        
        elif event.name == "handoff_requested":
            print(f"切换到: {event.item.target_agent}")
        
        elif event.name == "reasoning_item_created":
            print(f"推理: {event.item.content}")
    
    elif event.type == "raw_response_event":
        # 原始响应事件
        print(f"原始事件: {event.data.type}")
    
    elif event.type == "agent_updated_stream_event":
        # 代理更新事件
        print(f"新代理: {event.new_agent.name}")

def cancel(self) -> None

import asyncio

async def cancellable_execution():
    """可取消的执行"""
    
    result = await Runner.run_streamed(agent, input)
    
    # 在另一个任务中取消
    async def cancel_after(seconds: float):
        await asyncio.sleep(seconds)
        result.cancel()
        print("执行已取消")
    
    cancel_task = asyncio.create_task(cancel_after(10.0))
    
    try:
        async for event in result.stream_events():
            # 处理事件
            pass
    except asyncio.CancelledError:
        print("流式处理被取消")
    finally:
        cancel_task.cancel()

执行配置层
    └── RunConfig (执行配置)
        ├── ModelSettings (模型参数)
        ├── Guardrails (安全防护)
        └── Hooks (生命周期钩子)

执行结果层
    ├── RunResult (标准结果)
    └── RunResultStreaming (流式结果)
        └── StreamEvent (流式事件)

上下文层
    └── RunContextWrapper (上下文包装器)
        └── Usage (Token使用统计)

内部数据层
    ├── ModelInputData (模型输入数据)
    ├── CallModelData (模型调用数据)
    └── _ServerConversationTracker (服务器对话追踪)

classDiagram
    class RunConfig {
        +str | Model | None model
        +ModelProvider model_provider
        +ModelSettings | None model_settings
        +HandoffInputFilter | None handoff_input_filter
        +list[InputGuardrail] | None input_guardrails
        +list[OutputGuardrail] | None output_guardrails
        +int max_turns
        +bool trace_include_sensitive_data
        +CallModelInputFilter | None call_model_input_filter
        +SessionInputCallback | None session_input_callback
        +RunHooksBase | None hooks
        +str | None conversation_id
        +str | None previous_response_id
        
        +__init__(...)
    }
    
    class ModelSettings {
        +float | None temperature
        +int | None max_tokens
        +float | None top_p
        +float | None frequency_penalty
        +float | None presence_penalty
        +dict | None response_format
        +list[str] | None stop
        
        +merge(other: ModelSettings) ModelSettings
    }
    
    class InputGuardrail {
        <<Abstract>>
        +run(input: Any, context: Any)* InputGuardrailResult
    }
    
    class OutputGuardrail {
        <<Abstract>>
        +run(output: Any, context: Any)* OutputGuardrailResult
    }
    
    class RunHooksBase {
        <<Abstract>>
        +on_run_start(...)* None
        +on_run_end(...)* None
        +on_tool_call(...)* None
        +on_tool_result(...)* None
        +on_agent_switch(...)* None
    }
    
    class HandoffInputFilter {
        <<Callable>>
        (handoff_data) -> list[TResponseInputItem]
    }
    
    class CallModelInputFilter {
        <<Callable>>
        (call_model_data) -> ModelInputData
    }
    
    class SessionInputCallback {
        <<Callable>>
        (items) -> list[TResponseInputItem]
    }
    
    RunConfig --> ModelSettings : uses
    RunConfig --> InputGuardrail : contains
    RunConfig --> OutputGuardrail : contains
    RunConfig --> RunHooksBase : uses
    RunConfig --> HandoffInputFilter : uses
    RunConfig --> CallModelInputFilter : uses
    RunConfig --> SessionInputCallback : uses
    
    note for RunConfig "执行配置核心类<br/>控制模型、防护、钩子等<br/>所有参数都是可选的"
    note for ModelSettings "模型参数配置<br/>温度、token限制等<br/>支持配置合并"
    note for InputGuardrail "输入安全防护<br/>在执行前检查输入<br/>可触发tripwire中断执行"
    note for OutputGuardrail "输出安全防护<br/>在执行后检查输出<br/>可触发tripwire中断执行"