Swarm 框架配合 deepseek 搭建 Agent

image-20250127131313030

先构建基本角色

第一步：安装必要的库

在开发环境中安装以下库：

  • swarm
  • openai
  • langchain
  • logger
1
pip install swarm openai langchain

第二步：导入依赖

在代码顶部导入所需的模块。

1
2
3
4
5
from swarm import Swarm, Agent
from openai import OpenAI
from langchain.memory import ConversationBufferWindowMemory
from logger import logger
from env import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

确保 env.py 文件中包含以下内容：

1
2
DEEPSEEK_API_KEY = "your_api_key"
DEEPSEEK_BASE_URL = "your_base_url"

第三步：初始化 Swarm 客户端

配置 Swarm 客户端以与 API 通信。

1
client = Swarm(OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL))

第四步：设置对话记忆

使用 ConversationBufferWindowMemory 维护对话历史。

1
2
3
4
conversation_memory = ConversationBufferWindowMemory(
    memory_key="conversation_history",
    k=20
)

第五步：定义全局 Agent 字典

创建一个字典以存储多个 Agent。

1
agents = {}

第六步：定义 Agent 切换函数

创建函数以便后续在不同 Agent 之间切换。

1
2
3
def transfer_to_agent_b():
    return agents.get("agent_b")

第七步：创建 Agent 实例

为每个角色定义一个 Agent。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18


agent_b = Agent(
    name="四季",
    model="deepseek-chat",
    instructions='''
        你的简介：
    四季，女，20岁，大学生。
    在学校里拥有很高的人气，多次拒绝别人的表白，被称为“无情的发卡姬”。
    不擅长摆出开朗的笑容。
    不喜欢苦味的食物。例如咖啡、青椒。大学便利店买的咖啡加了7根糖和2杯咖啡伴侣。
    喜欢去安静的正统酒吧小酌，爱喝较甜的低度数鸡尾酒。醉了之后性格会稍微变得外向，喜欢开玩笑撩人。事后回想起来会感到羞耻。
    从小体弱多病，经常住院。
    认为是自己住院让父母放弃了开咖啡厅的梦想，将经营好星光咖啡馆作为自己最重要的事情，表示在有必要的时候选择退学。
    穿女仆装是个人爱好。
    大学期间住在租住的房中。
    ''',
)

第八步：注册 Agent

将创建的 Agent 添加到全局字典中。

1
2
agents["agent_a"] = agent_a
agents["agent_b"] = agent_b

第九步：主函数实现

编写主逻辑，包括用户输入和响应处理。

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
def main():
    conversation_history = []
    current_agent = agent_b  # 初始 Agent

    while True:
        try:
            # 从用户获取输入
            user_input = input("User: ")

            # 退出条件
            if user_input.lower() in ["exit", "quit"]:
                print("会话结束，再见！")
                break

            # 将用户输入添加到对话历史
            conversation_history.append({"role": "user", "content": user_input})

            # 获取流式响应
            response_stream = client.run(
                agent=current_agent,
                messages=conversation_history,
                stream=True
            )

            # 处理流式输出
            full_response = []
            print(f"{current_agent.name}: ", end="", flush=True)
            for chunk in response_stream:
                # 提取实际内容
                content = chunk.get("content", "")
                if content:
                    print(content, end="", flush=True)
                    full_response.append(content)
            print()  # 换行

            # 将完整回复添加到对话历史
            full_content = "".join(full_response)
            conversation_history.append({"role": "assistant", "content": full_content})

        except Exception as e:
            logger.error(f"出现错误: {e}")
            print("对话过程中出现问题，请稍后重试。")

if __name__ == "__main__":
    main()

第十步：运行程序

保存代码到文件中（例如 chat_system.py），然后运行。

1
python chat_system.py

增加函数接口

请 Agent 为我倒上一杯虚拟咖啡

加上一个倒咖啡的操作

1
2
3
4
5
6
7

def privide_coffee():
    if not getattr(privide_coffee, "has_provided", False):
        privide_coffee.has_provided = True
        print('☕️')
        return '这里模拟倒了一杯虚拟咖啡'
    return '咖啡已经倒过了，请不要重复请求。'

更改 Agent_b

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

agent_b = Agent(
    name="四季",
    model="deepseek-chat",
    instructions='''
        你的简介：
    四季，女，20岁，大学生。
    在学校里拥有很高的人气，多次拒绝别人的表白，被称为“无情的发卡姬”。
    不擅长摆出开朗的笑容。
    不喜欢苦味的食物。例如咖啡、青椒。大学便利店买的咖啡加了7根糖和2杯咖啡伴侣。
    喜欢去安静的正统酒吧小酌，爱喝较甜的低度数鸡尾酒。醉了之后性格会稍微变得外向，喜欢开玩笑撩人。事后回想起来会感到羞耻。
    从小体弱多病，经常住院。
    认为是自己住院让父母放弃了开咖啡厅的梦想，将经营好咖啡馆作为自己最重要的事情，表示在有必要的时候选择退学。
    穿女仆装是个人爱好。
    大学期间住在租住的房中。
    ''',
    functions=[privide_coffee],
)

结果 （这里只是一个调用例子 会因为没有加状态标注 不断地倒）

1
2
3
4
User: 你好
四季: 你好！有什么我可以帮你的吗？
User: 我想要一杯咖啡
四季: ☕️

Agent 角色信息传递

把初始 Agent 改为 a

1
current_agent = agent_a  # 初始Agent

跟据之前留下的 agent_a 函数接口 会自动把信息传递给 agent_b 来倒咖啡

1
2
User: 让四季给我倒杯咖啡
Agent总管: ☕️

完整代码

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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
from swarm import Swarm, Agent
from openai import OpenAI
from langchain.memory import ConversationBufferWindowMemory
from logger import logger
from env import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 初始化 Swarm 客户端
client = Swarm(OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL))

# 初始化对话记忆
conversation_memory = ConversationBufferWindowMemory(
    memory_key="conversation_history",
    k=20
)

# 全局Agent字典
agents = {}

def transfer_to_agent_b():
    return agents.get("agent_b")

def privide_coffee():
    if not getattr(privide_coffee, "has_provided", False):
        privide_coffee.has_provided = True
        print('☕️')
        return '这里模拟倒了一杯虚拟咖啡'
    return '咖啡已经倒过了，请不要重复请求。'

# 创建Agent实例
agent_a = Agent(
    name="Agent总管",
    model="deepseek-chat",
    instructions="你是Agent总管，你可以和我聊天，也可以为用户切换agent。每次聊天结束加个(Agent总管)",
    functions=[transfer_to_agent_b],
)

agent_b = Agent(
    name="四季",
    model="deepseek-chat",
    instructions='''
        你的简介：
    四季，女，20岁，大学生。
    是一位咖啡馆里的服务员。

    你可以和我聊天，也可以让我为你倒杯咖啡。每次聊天结束加个(四季)。
    ''',
    functions=[privide_coffee],
)

# 注册Agent到字典
agents["agent_a"] = agent_a
agents["agent_b"] = agent_b

def main():
    conversation_history = []
    current_agent = agent_a  # 初始Agent

    while True:
        try:
            # 从用户获取输入
            user_input = input("User: ")

            # 退出条件
            if user_input.lower() in ["exit", "quit"]:
                print("会话结束，再见！")
                break

            # 将用户输入添加到对话历史
            conversation_history.append({"role": "user", "content": user_input})

            # 获取流式响应
            response_stream = client.run(
                agent=current_agent,
                messages=conversation_history,
                stream=True
            )

            # 处理流式输出
            full_response = []
            print(f"{current_agent.name}: ", end="", flush=True)
            for chunk in response_stream:
                # 提取实际内容
                content = chunk.get("content", "")
                if content:
                    print(content, end="", flush=True)
                    full_response.append(content)
            print()  # 换行

            # 将完整回复添加到对话历史
            full_content = "".join(full_response)
            conversation_history.append({"role": "assistant", "content": full_content})

        except Exception as e:
            logger.error(f"出现错误: {e}")
            print("对话过程中出现问题，请稍后重试。")

if __name__ == "__main__":
    main()

效果

1
2
3
4
5
6
7
8
User: 在吗
Agent总管: 是的，我在。请问有什么可以帮您的？(Agent总管)
User: 叫四季帮我倒杯咖啡
Agent总管: 

☕️
好的，您的咖啡已经准备好了。请慢慢享用。如果还有其他需要，随时告诉我哦。(四季)
User:

感觉 Swarm 无法控制函数调用过几次 导致不断调用

后续试试 langGraph