基于 DeepSeek + OpenAI Agent 构建 Agent 实战

本文档基于 openai-agents 官方文档，详细介绍了一个支持异步调用的多智能体系统，实现了各个智能体的明确分工协作。
其实在我看来，这个框架的本质是：把 Swarm 改进成支持异步调用，并把手写 Agent 传递变成 Handoff 的封装，可能也做了增加稳定性的优化。

安装指南

使用以下命令安装依赖包：

1
pip install openai-agents

基础使用流程

系统使用包含以下两步：

  1. Agent 定义
  2. Agent 执行（输入用户问题）

基础示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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="base_url", 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)

执行后可能输出如下：

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

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

以上展示了系统的基本工作流程：Agent 定义、执行两个步骤。

复杂任务

在处理复杂任务时，系统可能面临以下技术挑战：

  1. 语言模型能力要求：需要同时具备推理规划和函数调用能力
  2. 提示词复杂度：需要同时处理规划制定、函数调用和结果输出
  3. 缺乏系统性规划：每次 Function Call 后重新思考，缺乏连贯的执行计划

（注：DeepSeek-R1 目前尚未支持函数调用功能）

解决方案：采用多智能体分工协作模式，将单一大型语言模型拆分为多个专门化的智能体。每个智能体负责特定任务，便于系统优化和评估。

我们遵循”智能体职责分离”原则，实现模块化设计，提高系统可维护性和可扩展性。

下面大致介绍一下工作流程：

系统工作流程示意：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
用户输入： "please show me the menu"
       │
       ▼
规划智能体【系统管理器】（Deepseek-R1）
  分析用户输入并确定调用策略：
  "调用智能体：菜单管理智能体
   用户原始输入：请展示菜单"
       │
       ▼
分配智能体【服务管理器】（Deepseek-V3）
  接收指令并执行：
  -> llm function call：show_order_menu
       │
       ▼
  获取菜单数据：
  -> 执行 show_order_menu 函数
       │
       ▼
语言处理智能体【多语言适配器】（Deepseek-V3，根据用户语言选择对应处理模块）
  -> 如英文翻译agent：english_translator
       │
       ▼
  返回处理后的菜单数据（包含处理模块标识）
       │
       ▼
  向用户返回处理结果

进阶实现示例

以下代码展示了多智能体协作系统的完整实现流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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="base_url", 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",
    instructions="你处于一个多Agent工作流中，职责是英文翻译，将收到任何文本（包括tool中）返回成英文结果。（请务必遵从以上原则，否则会破坏工作流）最后你要以你的身份来作为结尾告诉我这是来自谁的答复，如This response comes from the English translator.",
)

# 定义中文翻译Agent
chinese_translator = Agent(
    name="Chinese translator",
    model="deepseek-v3",
    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",
    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())

系统自检机制

系统实现了自检功能，通过在响应智能体中加入异常检测逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

agent_response = Agent(
    name="Agent 回复",
    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_response, 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

异常处理示例：

当出现异常如下文，系统将触发自检机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
总管输出： 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_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!

以上展示了基于”规划-执行”模式的多智能体系统实现：

通过将任务处理流程分解为规划层（系统管理器、服务管理器）和执行层（功能模块、语言处理模块），实现了智能体的职责分离，同时充分利用了 DeepSeek 等模型在推理方面的优势，规避了其在函数调用等方面的限制。系统还实现了自检和异常处理机制，提高了系统的可靠性和稳定性。