本文为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

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐