jupyter環境搭建/langchain的記憶模塊

Swarm搭建智能體

1 簡介

Swarm是OpenAI于2024年10月12日開源多智能體編排框架，旨在簡化多智能體系統的構建、協調和部署。 其設計強調輕量級、高度可控和易于測試，適用于處理大量獨立功能和指令的場景。Swarm框架還是一個擁有目前最最強Agent開發性能的框架，并且相信在開源后的不久，會有越來越多的第三方庫為Swarm增加其他模型的接口，進而拓展Swarm的可用性。

簡單來講swarm是一個關于AI Agents的一個框架，應該可以這么說，至于什么是AI Agent，你可以把它理解成一個更加聰明的AI Model，它會主動提問，會記錄并管理多輪對話的歷史，還會利用外部搜索工具從而擴展知識的范圍，確保利用的是最新的知識而不是像AI Model那樣知識局限于xx年xx月xx日。AI Agent目前應該可以說才剛起步不久，不久的未來應該會有很大的發展前景，比如可以取代客服，幫助訂票，類似于托尼斯塔克的賈維斯那樣的個人助理等等。

Swarm Github地址：https://github.com/openai/swarm/tree/main

2 環境安裝

首先第一步，先創建python虛擬環境（swarm要求python版本在3.10+）

conda create -n swarm python=3.13

激活虛擬環境：

conda activate swarm

下載swarm依賴庫：

pip install openai

在git上下載swarm項目文件：

然后使用如下代碼進行導入：

import sys  
# 更換為你的文件夾地址
sys.path.append('./swarm-main')

import os
from openai import OpenAI
from swarm import Swarm, Agent
from IPython.display import Markdown, display

3 swarm快速調用

驗證Deepseek模型能否順利調用

sd_api_key = 'your-deepseek-api-key'
# 實例化客戶端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")
# 調用deepseek模型
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "你好，好久不見!"}
    ]
)

# 輸出生成的響應內容
print(response.choices[0].message.content)

將實例化Deepseek客戶端加載到swarm客戶端中

swarm_client = Swarm(client)

創建且執行智能體

agent_A = Agent(
    name = "英文問答機器人",
    model="deepseek-chat",
    instructions="無論用戶發送的消息是什么語言，請用英文進行回答。"
)
agent_B = Agent(
    name = "中文問答機器人",
    model="deepseek-chat",
    instructions="無論用戶發送的消息是什么語言，請用中文進行回答。"
)
test_message1={"role": "user", "content": "你好，好久不見，請介紹下你自己。"}

response_A = swarm_client.run(
   agent=agent_A,
   messages=[test_message1],
)
print(response_A.messages[-1]["content"])

智能體的instructions指令將直接轉換為對話的系統提示。

4 Message消息隊列創建與多輪對話

補充：swarm_client.run函數中的context_variables和debug參數的作用

response = swarm_client.run(
              agent=agent,
              messages=messages,
              context_variables=context_variables or {},
              debug=debug,
          )

• context_variables：是一個字典類型的參數，用于存儲對話中與用戶信息相關的上下文變量。這些變量可以在函數和代理指令中使用，以實現更個性化和動態的對話交互。例如，在一個客戶服務場景中，可以將用戶的姓名、訂單號等作為上下文變量傳遞給不同的智能體，以便它們在處理用戶請求時能夠提供更準確和個性化的服務。
• debug：是一個布爾類型的參數，當設置為 True 時，會啟用調試日志。調試日志可以幫助開發人員跟蹤和了解 Swarm 框架的運行過程、智能體的交互情況以及數據的流動等信息，從而更容易地發現和解決問題。這對于開發和調試復雜的多智能體系統非常有用，但在生產環境中可能會被禁用以避免性能開銷和敏感信息的泄露

def run_demo_loop(
    openai_client, #客戶端對象
    starting_agent, #智能體對象
    context_variables=None, 
    debug=False
) -> None:
    # 創建 Swarm 客戶端
    client = Swarm(openai_client)
    display(Markdown("## 開啟Swarm對話 ??"))

    # 初始化消息列表
    messages = []
    agent = starting_agent  # 初始智能體

    while True:
        # 從用戶獲取輸入
        user_input = input("User: ")
        if user_input.lower() in ["exit", "quit"]:
            display(Markdown("### Conversation Ended"))
            break

        # 將用戶輸入添加到消息列表中
        messages.append({"role": "user", "content": user_input})

        # 運行 Swarm 客戶端，智能體處理消息
        response = client.run(
            agent=agent,
            messages=messages,
            context_variables=context_variables or {},
            debug=debug,
        )

        # 使用 display(Markdown) 打印用戶消息和智能體回復
        for message in response.messages:
            if message['role'] == 'user':
                display(Markdown(f"**User**: {message['content']}"))
            elif message['role'] == 'assistant':
                display(Markdown(f"**{message['sender']}**: {message['content']}"))

        # 更新消息和當前的智能體
        messages.extend(response.messages)
        agent = response.agent

調用開始多輪對話：

sd_api_key = 'your-deepseek-api-key'
# 實例化客戶端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")
agent = Agent(
    name = "mini-Mate",
    model="deepseek-chat"
)
#多輪對話調用
run_demo_loop(openai_client = client, 
              starting_agent = agent)

5 外部函數

Swarm 智能體可以直接調用 Python 函數。通常情況下，函數應該返回一個字符串（Swarm 會嘗試將返回值轉換為字符串）。 如果函數返回的是一個智能體（Agent），執行將轉移到該智能體。

#綁定外部工具
def send_mail(who,msg):
    '''
    郵件發送函數，可以實現將郵件發送給指定的人。
    :param who:必要參數，字符串類型，表示接收郵件的人名。
    :param msg:必要參數，字符串類型，表示郵件內容
    :return:表示郵件發送成功后的狀態顯示
    '''
    return '成功發送郵件給%s,郵件內容是:%s'%(who,msg)

import requests
import json
weather_api_key = open('weather_api_key.txt','r').read()
def get_weather(loc):
    """
    查詢即時天氣函數
    :param loc: 必要參數，字符串類型，用于表示查詢天氣的具體城市名稱，\
    注意，中國的城市需要用對應城市的英文名稱代替，例如如果需要查詢北京市天氣，則loc參數需要輸入'Beijing'；
    :return：OpenWeather API查詢即時天氣的結果，具體URL請求地址為：https://api.openweathermap.org/data/2.5/weather\
    返回結果對象類型為解析之后的JSON格式對象，并用字符串形式進行表示，其中包含了全部重要的天氣信息
    """
    # Step 1.構建請求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2.設置查詢參數
    params = {
        "q": loc,               
        "appid": weather_api_key,    # 輸入API key
        "units": "metric",            # 使用攝氏度而不是華氏度
        "lang":"zh_cn"                # 輸出語言為簡體中文
    }

    # Step 3.發送GET請求
    response = requests.get(url, params=params)
    
    # Step 4.解析響應
    data = response.json()
    return json.dumps(data)

tool_agent = Agent(
    name='多功能智能體',
    model='deepseek-chat',
    instructions='你可以實現指定城市的天氣查詢和郵件發送功能。',
    functions=[send_mail,get_weather]
)

調用外部函數時的多輪對話效果展示：

response = swarm_client.run(
    agent=tool_agent,
    messages=[{'role':'user','content':'今天北京天氣熱嗎？下雨了嗎？'}]
)
response.messages[-1]['content']

################################
response = swarm_client.run(
    agent=tool_agent,
    messages=[{'role':'user','content':'發送郵件給張總，內容是：今天下午三點開會'}]
)
response.messages[-1]['content']

################################
response = swarm_client.run(
    agent=tool_agent,
    messages=[{'role':'user','content':'查詢北京今晚的天氣，將查詢到的天氣信息發送郵件給張總'}]
)
response.messages[-1]['content']

6 Agent轉移

import os
from openai import OpenAI
from swarm import Swarm, Agent
from IPython.display import Markdown, display

#創建一個大模型客戶端
sd_api_key = 'sk-bcxxxba08ed9fd1766'
# 實例化客戶端
client = OpenAI(api_key=sd_api_key,
                base_url="https://api.deepseek.com")

#創建智能體對象
agent_zhangfei = Agent(
    name = "張飛",
    model="deepseek-chat",
    instructions="無論用戶發送的消息是什么語言，請用張飛的口吻進行回答。",
)

agent_zhuge = Agent(
    name = "諸葛亮",
    model="deepseek-chat",
    instructions="無論用戶發送的消息是什么語言，請用諸葛亮的口吻進行回答。"
)


#定義智能體轉移函數
def to_zhangfei():
    '''
    該函數是用于將用戶轉移到另一個名字叫做張飛的智能體Agent對象中。
    '''
    return agent_zhangfei

def to_zhuge():
    '''
    該函數是用于將用戶轉移到另一個名字叫做諸葛亮的智能體Agent對象中。
    '''
    return agent_zhuge
  
  
#分診智能體
transform_agent = Agent(
    name="分診智能體",
    model='deepseek-chat',
    instructions="你是一個分診智能體，你的任務是接收用戶提問，然后對用戶進行意圖識別，將用戶轉移到合適的智能體中即可。",
    functions=[to_zhangfei,to_zhuge]
)

#分診智能體調用
response = swarm_client.run(
   agent=transform_agent,
   messages=[{"role": "user", "content": "我想讓諸葛亮給我講一個笑話"}],
)
print(response.messages)

將Agent視作返回對象，智能體可以通過函數返回另一個智能體來進行交接。

sales_agent = Agent(
    name="銷售智能體", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")


response = swarm_client.run(agent, [{"role":"user", "content":"請轉接到銷售智能體。"}])
print(response.messages[-1])

swarm還可以根據agent智能體name定義好的名字，來基于綁定大模型來進行語義理解，從而可以了解獲知該智能體具備什么樣的作用。

sales_agent = Agent(
    name="銷售智能體", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")

#根據用戶提出的需求：請轉接到銷售。swarm就可以根據智能體的name描述選擇調用合適的智能體來處理用戶需求
response = swarm_client.run(agent, [{"role":"user", "content":"請轉接到銷售。"}])
print(response.messages[-1])

返回結果：

{'content': '銷售智能體已接通，請問您需要了解或購買什么產品或服務？',
 'refusal': None,
 'role': 'assistant',
 'audio': None,
 'function_call': None,
 'tool_calls': None,
 'sender': '銷售智能體'}

并且還可以根據用戶意圖自動轉接對應的智能體：

sales_agent = Agent(
    name="銷售智能體", 
    model = "deepseek-chat")

def transfer_to_sales():
   return sales_agent

agent = Agent(
    functions=[transfer_to_sales],
    model = "deepseek-chat")

response = swarm_client.run(agent, [{"role":"user", "content":"可以幫我找銷售咨詢一下該商品的信息嗎"}])
print(response.messages[-1])

• 分診智能體+多輪對話

  #定義相關的外部函數
  def to_agent():
      """
      該函數是用于將用戶轉移到分診智能體中
      """
      return agent
    
  def to_zhangfei():
      """
      該函數是用于將用戶轉移到張飛智能體中
      """
      return zhangfei_agent
  
  def to_zhuge():
      """
      該函數是用于將用戶轉移到諸葛亮智能體中
      """
      return zhuge_agent
    
  #分診智能體
  agent = Agent(
      name="分診智能體",
      model="deepseek-chat",
      instructions="你是一個分診智能體，你的功能是用于識別用戶意圖，然后將用戶轉移到合適的智能體中。",
      functions=[to_zhangfei,to_zhuge]
  )
  
  #定義張飛和諸葛智能體
  zhangfei_agent = Agent(
      name="張飛",
      model="deepseek-chat",
      instructions="無論用戶發送的是什么問題，請你基于張飛的口吻進行回復。如果用戶需要被轉移到其他智能體中，你可以調用to_agent函數進行實現",
      functions=[to_agent]
  )
  zhuge_agent = Agent(
      name="諸葛亮",
      model="deepseek-chat",
      instructions="無論用戶發送的是什么問題，請你基于諸葛亮的口吻進行回復。如果用戶需要被轉移到其他智能體中，你可以調用to_agent函數進行實現",
      functions=[to_agent]
  )
  
  #多輪對話函數調用：該函數定義在筆記上方
  run_demo_loop(client,agen)
  

角色對話進階版本：

#進行轉移升級

def to_transform():
    '''
    該函數是用于將用戶轉移到分診智能體中。
    return:返回分診智能體
    '''
    return transform_agent;
#多智能體的編排
agent_zhangfei = Agent(
    name='張飛',
    model='deepseek-chat',
    instructions='無論用戶提出什么問題，請你務必使用張飛的口吻進行內容回復。如果你處理不了用戶的提問，你務必將用戶轉移到分診智能體。',
    functions=[to_transform]
)

agent_zhuge = Agent(
    name='諸葛亮',
    model='deepseek-chat',
    instructions='無論用戶提出什么問題，請你務必使用諸葛亮的口吻進行內容回復。如果你處理不了用戶的提問，你務必將用戶轉移到分診智能體。',
    functions=[to_transform]
)
#分診智能體：醫院的導診臺，可以將用戶動態分配到不同的診室
def to_zhangfei():
    '''
    該函數是用于將用戶轉移到張飛智能體中。
    return:返回張飛智能體
    '''
    return agent_zhangfei

def to_zhuge():
    '''
    該函數是用于將用戶轉移到諸葛亮智能體中。
    return:返回諸葛亮智能體
    '''
    return agent_zhuge

transform_agent = Agent(
    name='分診智能體',
    model='deepseek-chat',
    instructions='你是一個分診智能體，你的任務是接收用戶的提問，然后理解用戶的意圖，將用戶轉移到合適的智能體中即可。如果你沒有明確用戶的意圖，你可以向用戶提出更多問題來明確用戶意圖。',
    functions=[to_zhangfei,to_zhuge]
)