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

pip install openai-agents

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)

《赴京举子》
万里风尘逐帝京，八方士子共云程。
墨香暗度关山月，剑气遥连北斗星。
金殿未登龙虎榜，青衫已染凤凰名。
何须更问功成事，醉卧长安笑公卿。

注：本诗采用李白豪放飘逸的笔法，描绘了举子千里赴京的宏大场景。诗中运用"墨香剑气"暗喻文韬武略，以"龙虎榜"与"凤凰名"展现功名与气节的对比，末联"醉卧长安"体现了对科举制度的超然态度，同时也反映了盛唐气象，延续了李白诗作中特有的浪漫气质与历史纵深感。

用户输入： "please show me the menu"
       │
       ▼
规划智能体【系统管理器】（Deepseek-R1，擅长推理规划）
  分析用户输入并确定调用策略：
  "调用智能体：菜单管理智能体
  用户原始输入：please show me the menu"
       │
       ▼
分配智能体【服务管理器】（Deepseek-V3-Tool，支持 Function Call）
  接收指令并执行：
  -> llm function call：show_order_menu
       │
       ▼
  获取菜单数据：
  -> 执行 show_order_menu 函数
       │
       ▼
语言处理智能体【多语言适配器】（Deepseek-V3-Tool，根据用户语言选择对应处理模块）
  -> 如英文翻译agent：english_translator
       │
       ▼
  返回处理后的菜单数据（包含处理模块标识）
       │
       ▼
  向用户返回处理结果

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

总管输出： Agent_planner->Agent_dispatcher: @MenuAgent @MenuTool: plz show me the menu
plz show me the menu

    1. 美式咖啡 ￥10
    2. 拿铁 ￥15
    3. 卡布奇诺 ￥15
    4. 摩卡 ￥20
    5. 热巧克力 ￥15
    6. 红茶 ￥10
    7. 松饼 ￥10
    8. 蛋糕 ￥15

结果输出： {'assistant': '中文翻译'} #这里由于幻觉走到了中文翻译，所以输出的是中文翻译，在反思agent中检测出来并优化重试
总管输出： Agent_planner->Agent_dispatcher: @MenuAgent @MenuTool: plz show me the current menu items and prices in English

    1. 美式咖啡 ￥10
    2. 拿铁 ￥15
    3. 卡布奇诺 ￥15
    4. 摩卡 ￥20
    5. 热巧克力 ￥15
    6. 红茶 ￥10
    7. 松饼 ￥10
    8. 蛋糕 ￥15

结果输出： You have displayed the following items to the customer:
1. Americano - ¥10
2. Latte - ¥15
3. Cappuccino - ¥15
4. Mocha - ¥20
5. Hot Chocolate - ¥15
6. Black Tea - ¥10
7. Muffin - ¥10
8. Cake - ¥15
This response comes from the English translator.
反思结果： The current menu items and prices in English are as follows:
1. Americano - ¥10
2. Latte - ¥15
3. Cappuccino - ¥15
4. Mocha - ¥20
5. Hot Chocolate - ¥15
6. Black Tea - ¥10
7. Muffin - ¥10
8. Cake - ¥15
Let me know if you need further assistance!