实战指南:LLM 决策与代码编排双路径)
openai-agents-python 多智能体编排Agent Orchestration实战指南LLM 决策与代码编排双路径【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python本指南基于 openai-agents-python 官方文档 multi_agent.md 展开系统讲解多智能体应用中最核心的两种编排方式让 LLM 自主决策下一步以及用代码显式控制智能体流转。读完本文你将掌握 Agents as Tools 与 Handoffs 两大 SDK 核心模式的选型与实现、代码化编排的四大经典套路结构化路由、链式、判官循环、并行化并了解它们在仓库中的真实示例与源码级实现细节。编排的本质谁来决定下一步编排Orchestration指的是智能体在应用中的流转方式哪些智能体运行、按什么顺序运行、下一步由谁决定。在 openai-agents-python 中编排主要分为两大路线让 LLM 做决策借助 LLM 的规划与推理能力让它根据任务自行决定采取什么步骤通过代码编排由你的程序显式决定智能体的流转顺序。这两种模式可以混用且各有取舍LLM 编排灵活、擅长开放式任务但结果不可完全预测代码编排确定性强在速度、成本与性能上更可控。下面分别深入讲解。方式一基于 LLM 的编排在 SDK 中一个 Agent 本质上是配备了指令instructions、工具tools与交接handoffs的 LLM。面对开放式任务时LLM 可以自主规划如何拆解任务用工具采取行动、获取数据用 handoffs 把子任务委派给子智能体。例如一个研究型智能体research agent可以被装备上如下能力Web search在线检索信息File search 与检索搜索专有数据与已连接的数据源Computer use在计算机上执行操作Code execution进行数据分析Handoffs交接给擅长规划、报告撰写等任务的专职智能体。两种核心 SDK 编排模式在 Python SDK 中LLM 编排最常见的两种模式对比如下引自 multi_agent.md模式工作原理适用场景Agents as tools智能体即工具一个 manager 智能体始终控制对话通过Agent.as_tool()调用专职智能体希望由一个智能体负责最终答案、整合多个专职智能体的输出或在单一位置统一施加 SDK 护栏guardrailsHandoffs交接triage 智能体把对话路由给专职智能体专职智能体成为该回合剩余部分的活跃智能体希望专职智能体直接回复、保持提示词聚焦或希望交接切换活跃指令而不需要 manager 转述结果选型建议当专职智能体只应协助一个有边界的子任务、不应接管面向用户的对话时使用agents as tools当路由本身是工作流的一部分、希望被选中的专职智能体接管当前回合的剩余部分时使用handoffs。两种模式也可以组合一个 triage 智能体可以先 handoff 给专职智能体而该专职智能体内部仍可把其他智能体作为工具调用以完成更窄的子任务。LLM 编排适合开放式任务但要取得好效果官方文档给出了 5 条关键战术投资好的提示词明确说明有哪些工具、如何使用、智能体必须遵守哪些约束监控并迭代观察哪里出错持续迭代提示词允许智能体自我反思与改进例如放进循环中让它自我批评或把错误信息回喂给它改进使用专精单一任务的专职智能体优于一个期望什么都行的通用智能体投资评估evals用评测来驱动智能体持续变好。如果你需要这套编排风格背后的核心 SDK 原语可以从 tools、handoffs 和 running agents 开始。Agents as Tools让 manager 智能体掌控全局智能体即工具的思想是让一个中央智能体编排一组专职智能体而不是把控制权交出去。实现方式是把子智能体通过Agent.as_tool()建模成工具。as_tool是Agent上的一个方法其完整签名支持以下参数tool_name/tool_description工具名与描述不传时工具名由智能体名转换而来max_turns、run_config、hooks嵌套运行的常见运行期选项previous_response_id、conversation_id、session嵌套运行的状态选项默认不继承父运行的会话状态如需共享需显式传入同一个sessionneeds_approval是否在调用前暂停等待人工审批parameters、input_builder、include_input_schema结构化输入支持custom_output_extractor自定义嵌套运行的输出提取is_enabled运行期条件启用/禁用on_stream监听嵌套智能体的流式事件。最经典的翻译示例完整可运行版本见 agents_as_tools.pyimport asyncio from agents import Agent, Runner spanish_agent Agent( nameSpanish agent, instructionsYou translate the users message to Spanish, ) french_agent Agent( nameFrench agent, instructionsYou translate the users message to French, ) orchestrator_agent Agent( nameorchestrator_agent, instructions( You are a translation agent. You use the tools given to you to translate. If asked for multiple translations, you call the relevant tools. ), tools[ spanish_agent.as_tool( tool_nametranslate_to_spanish, tool_descriptionTranslate the users message to Spanish, ), french_agent.as_tool( tool_nametranslate_to_french, tool_descriptionTranslate the users message to French, ), ], ) async def main(): result await Runner.run(orchestrator_agent, inputSay Hello, how are you? in Spanish.) print(result.final_output) if __name__ __main__: asyncio.run(main())结构化输入默认情况下Agent.as_tool()期望一个只含input字符串字段的对象{input: ...}但你也可以通过parameters传入一个 Pydantic 模型或 dataclass 类型来暴露结构化 schemainclude_input_schemaTrue在生成的嵌套输入中附带完整 JSON Schemainput_builder...完全自定义结构化工具参数如何变成嵌套智能体的输入RunContextWrapper.tool_input嵌套运行上下文中包含解析后的结构化载荷。完整可运行示例见 agents_as_tools_structured.py。条件启用is_enabledas_tool的is_enabled参数支持布尔值、同步函数与异步函数接收(context, agent)并返回布尔值用于在运行期动态过滤 LLM 可见的工具。禁用工具在运行期对 LLM完全隐藏非常适合按请求作用域控制能力可见性按环境dev vs prod控制工具可用性A/B 测试不同工具配置基于运行状态做动态工具过滤。注意is_enabled控制可见性与分发但不能替代依赖工具参数或访问资源的授权检查——这类检查应放在工具实现内部或使用工具输入护栏tool input guardrails与审批approvals。MCP 服务器也必须自行授权受保护的操作。完整示例见 agents_as_tools_conditional.py其中演示了如何根据用户语言偏好动态启用法语/意大利语响应工具。审批门与自定义输出Agent.as_tool(..., needs_approval...)与function_tool共用同一套审批流程需要审批时运行会暂停待审批项出现在result.interruptions中随后通过result.to_state()结合state.approve(...)/state.reject(...)恢复运行完整模式见 human_in_the_loop.md。custom_output_extractor则允许你在结果返回中央智能体之前改写输出例如从子智能体的聊天历史中提取 JSON 载荷、把 Markdown 转成纯文本或 CSV、或在响应缺失/畸形时提供回退值。嵌套的RunResult还暴露agent_tool_invocation元数据外层工具名、调用 ID、原始参数详见 results.md。流式嵌套运行传入on_stream回调即可监听嵌套智能体的流式事件流结束后仍返回其最终输出。事件类型与StreamEvent[type]对应raw_response_event、run_item_stream_event、agent_updated_stream_event提供on_stream会自动以流式模式运行嵌套智能体并排空流。同步或异步回调均可见 agents_as_tools_streaming.py。Handoffs把对话交接给专职智能体Handoffs 让一个智能体把任务委派给另一个智能体特别适合不同智能体各自专精不同领域的场景——例如客服应用里分别处理订单状态、退款、FAQ 的智能体。Handoffs 对 LLM 而言表示为工具若存在名为Refund Agent的交接目标工具名就是transfer_to_refund_agent。创建交接非常简单Agent的handoffs参数既可以接收Agent实例也可以接收自定义的Handoff对象。直接传Agent时其handoff_description若设置会追加到默认工具描述之后用于提示模型何时选择该交接。from agents import Agent, handoff billing_agent Agent(nameBilling agent) refund_agent Agent(nameRefund agent) triage_agent Agent(nameTriage agent, handoffs[billing_agent, handoff(refund_agent)])handoff() 函数的自定义能力handoff()是 SDK 提供的交接工厂函数其签名见 src/agents/handoffs/init.py支持以下参数agent交接目标智能体tool_name_override默认工具名为transfer_to_agent_name由Handoff.default_tool_name()解析可覆盖tool_description_override覆盖默认工具描述on_handoff交接被调用时执行的回调可用于提前触发数据拉取等副作用可接收input_type控制的 LLM 生成输入input_type交接工具调用参数tool-call arguments的 schemainput_filter过滤下一个智能体接收到的输入is_enabled布尔值或函数用于运行期动态启用/禁用交接nest_handoff_history可选针对单个交接覆盖 RunConfig 层的nest_handoff_history设置。from agents import Agent, handoff, RunContextWrapper def on_handoff(ctx: RunContextWrapper[None]): print(Handoff called) agent Agent(nameMy agent) handoff_obj handoff( agentagent, on_handoffon_handoff, tool_name_overridecustom_handoff_tool, tool_description_overrideCustom description, )注意handoff()辅助函数始终把控制权转移给你传入的特定agent。如果有多个可能的目标请为每个目标注册一个 handoff 让模型自行选择只有当你的自有代码必须在调用时决定返回哪个智能体时才需要自定义Handoff。Handoff 输入input_type某些场景希望 LLM 在调用交接时提供一些数据例如交接给 Escalation agent 时让模型给出原因以便记录from pydantic import BaseModel from agents import Agent, handoff, RunContextWrapper class EscalationData(BaseModel): reason: str async def on_handoff(ctx: RunContextWrapper[None], input_data: EscalationData): print(fEscalation agent called with reason: {input_data.reason}) agent Agent(nameEscalation agent) handoff_obj handoff( agentagent, on_handoffon_handoff, input_typeEscalationData, )input_type描述的是交接工具调用本身的参数 schemaSDK 会把它暴露给模型作为交接工具的parameters、本地校验返回的 JSON并把解析后的值传给on_handoff。使用要点它不替换下一个智能体的主输入也不改变交接目标——接收智能体仍能看到对话历史除非用input_filter或嵌套交接历史设置修改它和RunContextWrapper.context是两回事input_type用于模型在交接时决定的元数据如reason、language、priority、summary应用状态与依赖应放入contextis_enabled在模型返回交接参数之前就被评估因此不能用来授权参数内含的值——若授权依赖解析后的字段应在on_handoff开头检查并在失败时 raise。输入过滤器Input Filters交接发生时新智能体接管对话并默认能看到全部历史。若想改变这一点可用input_filter它是一个接收HandoffInputData并返回新HandoffInputData的函数。HandoffInputData包含input_historyRunner.run(...)开始前的输入历史pre_handoff_items交接被调用的智能体回合之前产生的条目new_items当前回合产生的条目含交接调用与交接输出input_items可选的、转发给下一个智能体的条目替代new_items可在保持new_items完整用于会话历史的同时过滤模型输入run_context交接调用时的活跃RunContextWrapper。常用模式如从历史中移除所有工具调用已内置在agents.extensions.handoff_filtersfrom agents import Agent, handoff from agents.extensions import handoff_filters agent Agent(nameFAQ agent) handoff_obj handoff( agentagent, input_filterhandoff_filters.remove_all_tools, )SDK 还提供 beta 级的RunConfig.nest_handoff_history嵌套交接历史支持默认关闭可用handoff_history_mapper自定义映射单个交接可用handoff(...)的nest_handoff_historyTrue/False覆盖。另注意handoffs 局限在单次 run 内输入护栏只作用于链路中的第一个智能体输出护栏只作用于产生最终输出的智能体。推荐提示词为确保 LLM 正确理解 handoffs官方推荐在提示词中包含交接说明使用agents.extensions.handoff_prompt.RECOMMENDED_PROMPT_PREFIX或调用prompt_with_handoff_instructions自动添加from agents import Agent from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX billing_agent Agent( nameBilling agent, instructionsf{RECOMMENDED_PROMPT_PREFIX} Fill in the rest of your prompt here., )完整的路由式交接示例triage 智能体按语言交接给法语/西班牙语/英语智能体支持流式输出与多轮对话见 routing.py。方式二基于代码的编排LLM 编排固然强大但代码编排在速度、成本与性能上更确定、更可预测。官方文档归纳了四种常见套路结构化输出路由用 structured outputs 生成结构良好的数据供代码检查——例如让智能体把任务分类到几个类别中再按类别选择下一个智能体链式处理把上一个智能体的输出转换为下一个的输入将写博客这类任务拆成调研 → 写提纲 → 写正文 → 批评 → 改进的步骤序列判官循环在while循环中每次运行任务智能体产出结果再运行评估智能体打分反馈直到评估通过为止并行执行用asyncio.gather等 Python 原语并行运行多个互不依赖的智能体显著提速。仓库在 examples/agent_patterns 目录提供了全部可运行示例。结构化输出 确定性路由deterministic.py 演示了纯代码编排的确定性流程第一步智能体生成故事提纲第二步检查智能体用output_typeOutlineCheckerOutputPydantic 模型含good_quality、is_scifi两个布尔字段输出结构化判定代码检查判定结果决定是否继续写故事——这正对应用结构化输出做门控路由的思路class OutlineCheckerOutput(BaseModel): good_quality: bool is_scifi: bool outline_checker_agent Agent( nameoutline_checker_agent, instructionsRead the given story outline, and judge the quality. Also, determine if it is a scifi story., output_typeOutlineCheckerOutput, )随后代码显式把关卡闸门if not outline_checker_result.final_output.good_quality: ...决定是否进入写故事环节。整个流程包在一个with trace(Deterministic story flow):中保证端到端可追踪。LLM as a Judge判官循环llm_as_a_judge.py 实现生成 → 评估 → 反馈 → 再生成的循环story_outline_generator产出提纲evaluator智能体以EvaluationFeedbackdataclass含feedback与score: pass | needs_improvement | fail为结构化输出进行评判分数不是pass时把反馈追加进input_items重新生成直到通过dataclass class EvaluationFeedback: feedback: str score: Literal[pass, needs_improvement, fail] evaluator AgentNone这种任务智能体 评估智能体的双角色循环是提升输出质量的通用手段也可以用小模型做生成、大模型做评估以优化成本。并行化asyncio.gather 提速parallelization.py 展示了并行模式用asyncio.gather同时运行三次西班牙语翻译智能体再用一个translation_picker智能体从多个候选中挑选最佳翻译。注意所有调用都放在同一个trace(Parallel translation)上下文内整条工作流在追踪系统中保持为一个完整 trace。凡是互不依赖的任务多路检索、多方案生成、批量子任务都可以用 Python 原生的并发原语直接并行。强制工具使用与护栏与编排配套的常用技巧还有用ModelSettings(tool_choicerequired)强制模型必须调用工具见 forcing_tool_use.py以及为并行化配套的输入/输出护栏tripwire 机制见 input_guardrails.py 与 output_guardrails.py。两种方式的权衡与混合使用总结取舍LLM 编排把下一步做什么交给模型适合开放式、探索性任务代价是结果不确定、需要靠提示词迭代与 evals 收敛质量代码编排把流程写死在程序中换来可预测的速度、成本与行为适合流程固定、质量门槛明确的场景如发布管线、合规流程、批量处理。实践中两者常组合出现代码负责外层的确定性骨架何时开始、何时停止、并行度LLM 负责骨架内的智能决策如何拆解、选哪个工具、交接给谁。例如 triage 智能体负责路由LLM 决策而路由到哪个专职智能体后的处理流程由代码控制。SDK 的 tracing 能力with trace(...)让混合编排的每一步都可观测、可排查是生产落地多智能体应用的重要一环。相关指南Agents组合模式与智能体配置ToolsAgent.as_tool()与 manager 风格编排的完整 APIHandoffs专职智能体之间的委派Running agents单次运行的控制与对话状态Quickstart一个最小化的端到端 handoff 示例examples/agent_patterns本文涉及全部模式的可运行示例集合。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考