OpenAI SDK 智能体 (Agentic AI) 的 handoff (任务转交) 和 as_tool(智能体即工具) 的使用
OpenAI SDK 智能体 (Agentic AI) 的 handoff (任务转交) 和 as_tool(智能体即工具) 的使用
本文为Agentic AI所需的大 模型 API调用 的一些API示范,注重于使用SCNet以及DeepSeek的基于OpenAI 以及 OpenAI SDK 的 API调用。本文为公益类代码,由DeepSeek辅助生成,经过实例测试。
有关SCNet和DeepSeek API的调用,见原文
https://blog.csdn.net/YucongCai/article/details/159774133
在OpenAI Agents SDK 和目前主流的Agentic AI系统,在多Agents 的情况下,常常会遇到任务交接或权限交接的需求,这就是 handoff 的目的。在OpenAI SDK 中,agent 在定义之初或之后,可以指定一个或几个handoff agents,以此让agent将任务在特定状态将任务和权限装交给被指定的agent。例如, agent a 的 handoff 如果被指定为 agent b和 agent c,那么agent a就可以在满足条件时根据定义将任务转交给agent b 或agent c中的一个agent,而收到任务的agent 就会接替agent a去尝试完成任务同时获取agent a的权限。要注意,一个agent只能将其任务和权限handoff 给一个agent。而且,权限在这里指的是agent a所控制的任务状态和workflow,agents 可以使用的工具仍被各自定义,不会被自动转接。可以理解成一个员工将任务交给了另一个员工,但员工各自仍是独立的个体,而guardrail等系统设定是独立定义的。
而OpenAI SDK的 agent(s) astool是另一个概念。在OpenAI SDK中,agent可以通过 as_tool function 被转化为工具,以此被其他其它agent(s)调用。
需要注意的是,一个agent仅可将任务handoff给一个agent,然而一个agent可以同时使用多个agents as_tool 并行使用。
接下来,本文会展示两组源代码,读者可以此测试相关性能。Agentic AI的SDK仍处于发展阶段,相关概念比代码本身更为重要。其实,有需求的读者可以根据需求自主编写代码处理history/session,agents handoff,as_tool 等,只不过使用OpenAI SDK相对而言方便一些,对于简单系统可以降低成本。
handoff
import asyncio
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI
from agents import (
Agent, Runner,
OpenAIChatCompletionsModel, set_tracing_disabled
)
# --- Setup ---
load_dotenv(override=True)
set_tracing_disabled(True) # optional: disable OpenAI tracing
client = AsyncOpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1"
)
model = OpenAIChatCompletionsModel(
model="deepseek-chat",
openai_client=client
)
# --- Specialist agents ---
billing_agent = Agent(
name="billing_agent",
instructions="You are a billing expert. Answer questions about invoices, payments, and charges.",
model=model
)
refund_agent = Agent(
name="refund_agent",
instructions="You are a refund expert. Handle all refund requests and policy questions.",
model=model
)
# --- Triage agent (handoff, no session) ---
triage_agent = Agent(
name="triage_agent",
instructions=(
"You are a customer service triage agent. "
"If the user asks about a bill or payment, hand off to billing_agent. "
"If they ask for a refund, hand off to refund_agent."
),
handoffs=[billing_agent, refund_agent],
model=model
)
# --- Run without any session ---
async def main():
# This will automatically hand off to billing_agent
result = await Runner.run(
triage_agent,
"I was charged twice for my last subscription. Can you help?"
# no session parameter → no history persistence
)
print(result.final_output)
if __name__ == "__main__":
# asyncio.run(main())
await main()
as_tool (并行的示例)
import asyncio
import os
from dotenv import load_dotenv
from openai import AsyncOpenAI
from agents import (
Agent, Runner, SQLiteSession,
OpenAIChatCompletionsModel, set_tracing_disabled,
ModelSettings # needed for parallel tool calls
)
# -------------------- Environment & SDK Setup --------------------
load_dotenv(override=True)
set_tracing_disabled(True) # disable OpenAI tracing (optional)
# 1. Configure DeepSeek client (OpenAI‑compatible)
client = AsyncOpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1"
)
# 2. Model wrapper – DeepSeek supports tool calling
model = OpenAIChatCompletionsModel(
model="deepseek-chat",
openai_client=client
)
# -------------------- Define Specialist Agents --------------------
billing_agent = Agent(
name="billing_agent",
instructions="You are a billing expert. Answer questions about invoices, payments, and charges.",
model=model
)
refund_agent = Agent(
name="refund_agent",
instructions="You are a refund expert. Handle all refund requests and policy questions.",
model=model
)
# -------------------- Triage Agent (Handoff) --------------------
triage_agent = Agent(
name="triage_agent",
instructions=(
"You are a customer service triage agent. "
"If the user asks about a bill or payment, hand off to billing_agent. "
"If they ask for a refund, hand off to refund_agent."
),
handoffs=[billing_agent, refund_agent],
model=model
)
# -------------------- Agents for Parallel Tool Calls --------------------
# These will be wrapped as tools and called in parallel by the manager
finance_agent = Agent(
name="Finance_Analyst",
instructions="Analyze the financial health of a given company (revenue, profit, debt).",
model=model
)
market_agent = Agent(
name="Market_Analyst",
instructions="Analyze the market position and key competitors of a given company.",
model=model
)
# Wrap them as tools
finance_tool = finance_agent.as_tool(
tool_name="get_financial_analysis",
tool_description="Get a financial analysis of a company."
)
market_tool = market_agent.as_tool(
tool_name="get_market_analysis",
tool_description="Get a market analysis of a company."
)
# Manager agent that calls both tools in parallel
manager_agent = Agent(
name="Research_Manager",
instructions=(
"You are a research manager. For any company, you MUST call BOTH "
"get_financial_analysis and get_market_analysis in parallel to provide a complete report."
),
tools=[finance_tool, market_tool],
model=model,
model_settings=ModelSettings(parallel_tool_calls=True) # enables LLM‑driven parallelism
)
# -------------------- Optional: Session for multi‑turn --------------------
session = SQLiteSession("demo_user", "conversations.db")
# -------------------- Main Async Runner --------------------
async def main():
print("=== Demo 1: Handoff (triage → billing) ===\n")
result_handoff = await Runner.run(
triage_agent,
"I was charged twice for my last subscription. Can you help?",
session=session # optional, keeps conversation history
)
print(f"Handoff result:\n{result_handoff.final_output}\n")
print("=== Demo 2: Agent‑as‑Tool (parallel calls) ===\n")
result_parallel = await Runner.run(
manager_agent,
"Give me a full analysis of Tesla (TSLA).",
session=session
)
print(f"Parallel tools result:\n{result_parallel.final_output}\n")
if __name__ == "__main__":
await main()
我在找工作,HR或项目合作请联系:yucongcai_business@outlook.com
与科研相关的请联系:yucongcai_research@outlook.com
更多推荐




所有评论(0)