Agentic AI 干货！DeepSeek + OpenAI SDK 构建 Agent 实战

from agents import Agent, Runner, set_default_openai_client, set_default_openai_api, set_tracing_disabled
from openai import AsyncOpenAI

# 配置 七牛 API（请替换为实际的 API Key）
custom_client = AsyncOpenAI(base_url="https://api.qnaigc.com/v1", api_key="sk-xxxxx")
set_default_openai_client(custom_client)
# 下面两行很重要，含泪踩过的坑！
set_default_openai_api("chat_completions")
set_tracing_disabled(True) 

# Agent 定义示例
libai_agent = Agent(
    name="libai",
    model="deepseek-r1",  # 使用实际部署的模型名称
    instructions="模拟李白风格，根据用户输入创作诗歌。"
)

# 执行 Agent 并传入查询
result = Runner.run_sync(libai_agent, "请创作一首描写科举考生赴京场景的诗。")
print(result.final_output)

import asyncio
from agents import Agent, Runner, set_default_openai_client, set_default_openai_api, set_tracing_disabled, function_tool
from openai import AsyncOpenAI

custom_client = AsyncOpenAI(base_url="https://api.qnaigc.com/v1", api_key="sk-xxxxx")
set_default_openai_client(custom_client)
set_default_openai_api("chat_completions")
set_tracing_disabled(True)


# 定义英文翻译Agent
english_translator = Agent(
    name="English translator",
    model="deepseek-v3-tool",
    instructions="你处于一个多Agent工作流中，职责是英文翻译，将收到任何文本（包括tool中）返回成英文结果。（请务必遵从以上原则，否则会破坏工作流）最后你要以你的身份来作为结尾告诉我这是来自谁的答复，如This response comes from the English translator.",
)

# 定义中文翻译Agent
chinese_translator = Agent(
    name="Chinese translator",
    model="deepseek-v3-tool",
    instructions="你作为中文翻译，将收到任何文本返回成中文结果。你要以你的身份来作为结尾告诉我来自谁的答复",
)

# 定义菜单展示工具
@function_tool
def show_order_menu():
    menu='''
    1. 美式咖啡 ￥10
    2. 拿铁 ￥15
    3. 卡布奇诺 ￥15
    4. 摩卡 ￥20
    5. 热巧克力 ￥15
    6. 红茶 ￥10
    7. 松饼 ￥10
    8. 蛋糕 ￥15
    '''
    print(menu)
    return "你向顾客展示了"+menu

# 定义执行 Agent
agent_dispatcher = Agent(
    name="dispatcher",
    model="deepseek-v3-tool",
    instructions="你是副总管，请根据用户的问题调用相关工具，拿到相关工具结果后，再根据用户的文本语言选择合适的翻译Agent帮助你回复",
    handoffs=[chinese_translator,english_translator],
    tools=[show_order_menu]
)

# 定义规划 Agent，只负责根据用户问题分配任务，不直接回答问题
agent_planner = Agent(
    name="planner",
    model="deepseek-r1",
    instructions=(
        "你现在处于多智能体协作流程中，身份是 Agent 总管。你的任务是："
        "根据用户的问题分析并决定调用哪种功能，告诉副总管应该调用哪个Agent或tool处理用户的问题，最后加入用户原文"
        "仅输出该指令文本，不包含任何具体答案。示例如：Agent_planner->Agent_dispatcher: @MenuAgent @MenuTool: plz show me the menu"
    ),
)
async def main():
    reply=None
    # 用户提问
    question = "plz show me the menu"

    # 总管处理用户输入，输出分配指令
    orderFromPlanner = await Runner.run(agent_planner, question)
    print("planner输出：", orderFromPlanner.final_output)
        
    # 副总管根据总管指令调用工具并进行翻译
    result = await Runner.run(agent_dispatcher, input=orderFromPlanner.final_output)
    print("dispatcher输出：", result.final_output)


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

agent_reflect = Agent(
    name="reflect",
    model="deepseek-r1",
    instructions=(
        "你现在负责整理多智能体的输出，生成最终回复。"
        "输入包含各 Agent 的输出和用户问题。"
        "如果发现流程异常，有两种情况："
        "1. 可通过重新生成问题并执行流程解决，请在开头输出RETRY加重试次数1（若原文中已包含重试次数，则重试次数+1）便于程序做判断，接下来你的输出会作为新的用户输入，请你根据用户问题（保留原文原意原语音）重新生成易于命中的问题。"
        "2. 工作流或重试次数超过3次，请在开头输出ERROR，并在接下来分析原因。"
    ),
)

async def main():
    response = None
    # 用户输入
    query = "plz show me the menu "

    while True:
        
        # 总管处理用户输入，输出分配指令
        orderFromPlanner = await Runner.run(agent_planner, query)
        print("总管输出：", orderFromPlanner.final_output)
            
        # 副总管根据总管指令调用工具并进行翻译
        result = await Runner.run(agent_dispatcher, input=orderFromPlanner.final_output)
        print("结果输出：", result.final_output)
        

        # 响应管理器生成最终输出
        response = await Runner.run(agent_reflect, input="用户输入:" + query + "\n系统指令:" + orderFromPlanner.final_output + "\n执行结果:" + result.final_output)
        
        if response.final_output.startswith("RETRY"):
            print("重试")
            query = response.final_output
        else:
            print("反思结果：", response.final_output)
            break