【LLM】用SWARM构建编排层：一个电信用例

在当今快速发展的人工智能环境中，协调多个人工智能代理无缝协作变得越来越重要。OpenAI最近发布的SWARM框架为创建在不同代理之间委托任务的编排层提供了一个简单而强大的解决方案。在这篇文章中，我们将通过电信行业的一个实际例子来探索如何使用SWARM构建这样一个层。

SWARM是什么？

SWARM是一个实验框架，旨在管理多个AI代理之间的复杂交互。其主要特征包括：

1. 基于代理的体系结构：SWARM允许您创建专用代理，每个代理都有自己的一组指令和可用功能（工具）。
2. 动态交互循环：该框架管理代理交互、函数调用和代理之间潜在切换的连续循环。
3. 无状态设计：SWARM在调用之间是无状态的，提供了对编排过程的透明度和细粒度控制。
4. 直接函数调用：代理可以直接调用Python函数，增强其功能并与现有系统集成。
5. 上下文管理：SWARM提供上下文变量，用于维护跨代理交互的状态。
6. 灵活的切换：代理可以根据需要动态地将控制权切换到其他专业代理。
7. 实时交互：该框架支持流式响应以获得即时反馈。
8. 模型无关：SWARM适用于任何与OpenAI兼容的客户端，包括在Hugging Face TGI或vLLM等平台上托管的模型。

为什么选择SWARM？

作为一名人工智能从业者，我开始重视简单性和快速实施。我的哲学是：

💡 如果你能在几个小时的专注深度工作中把一个抽象的概念变为现实，这是值得探索的。💡

我们经常陷入复杂的框架和冗长的设置中。但是，如果最有影响力的想法可以通过一个简单的colab笔记本快速测试呢？

这种方法要求我们去除不必要的复杂性。这是关于寻找工具和方法，让我们迅速从概念转变为原型。通过接受这种思维方式，我们为创新打开了大门，使人工智能开发更容易获得。

哲学够了。让我们做真正的工作。

构建电信编排层

让我们通过一个示例来了解如何使用SWARM为电信公司构建编排层。

👨‍💻 完整的代码可以在这里找到：https://github.com/goodrahstar/orchestration_layer/blob/main/Orchestrat…

1.定义您的代理

在我们的电信编排层，我们创建了专门的代理来处理各种任务。每个代理都配备了特定的功能和知识，使他们能够有效地解决其指定的专业领域。

• -分类代理：确定哪个代理最适合回答用户的查询。
• -OLT代理：处理与光线路终端（OLT）数据相关的查询。
• -CRM代理：从CRM系统中检索客户信息。
• -ACS代理：管理与自动配置服务器（ACS）数据相关的查询。
• -解决方案代理：与知识图集成以解决客户问题。

2.实施数据连接器（工具）

对于每个代理，让我们定义他们可以使用的工具。例如，以下是从不同来源加载数据的工具。这些工具可以替换为不同类型的工具，如文本到SQL、API调用或从日志文件读取数据。

# Load data from various sources
def load_acs_data():
   df = pd.read_csv("SampleACS.csv")
   df['Customer ID'] = df['Customer ID'].astype(str)
   return df
def load_olt_data_dictionary():
   return pd.read_excel("OLT Data Dictionary.xlsx")
def load_crm_data():
   df = pd.read_csv("CRM dataset.csv")
   df['Customer ID'] = df['Customer ID'].astype(str)
   return df
def load_olt_status_and_alarms():
   df = pd.read_excel("OLT status and alarms.xlsx")
   return df
def load_ticketing_data():
   df = pd.read_excel("Ticketing sample data.xlsx")
   df['CustomerAccount'] = df['CustomerAccount'].astype(str)
   return df

# Integrated customer data
class IntegratedCustomerData:
   def __init__(self):
       self.acs_data = load_acs_data()
       self.crm_data = load_crm_data()
       self.olt_status = load_olt_status_and_alarms()
       self.ticketing_data = load_ticketing_data()
   def get_customer_info(self, customer_id):
       customer_info = {}
       
       if customer_id in self.crm_data['Customer ID'].values:
           customer_info['CRM'] = self.crm_data[self.crm_data['Customer ID'] 
           == customer_id].to_dict('records')[0]
       
       if customer_id in self.acs_data['Customer ID'].values:
           customer_info['ACS'] = self.acs_data[self.acs_data['Customer ID'] 
           == customer_id].to_dict('records')[0]
       
       if customer_id in self.olt_status['olt_id'].values:
           customer_info['OLT_Status'] = self.olt_status[self.olt_status['olt_id'] == customer_id].to_dict('records')[0]
       
       if customer_id in self.ticketing_data['CustomerAccount'].values:
           customer_info['Tickets'] = self.ticketing_data[self.ticketing_data['CustomerAccount'] 
           == customer_id].to_dict('records')
       
       return customer_info
integrated_data = IntegratedCustomerData()

我添加了一个额外的工具，用于与查询拥抱脸空间的知识图进行交互。这使我们的代理能够利用庞大的结构化信息库，增强他们处理和理解复杂信息的能力。

def knowledgegraph_tool(query):
 from gradio_client import Client
 client = Client("neuronslabs/Telcom_GraphRAG_Agent", hf_token)
 result = client.predict(
     history=[[query, None]],
     api_name="/bot"
 )
 return(result[-1][-1])

👉 Learn how i build Knowledge Graphs here : 

https://www.youtube.com/watch?v=5XTshvMd3Rk

3.定义Agent功能

现在让我们定义代理函数。这些函数将使用我们将在后续步骤中定义的代理来调用。他们将负责在后台调用之前定义的工具，并执行必要的操作，如数据检索或内容生成。

# Agent functions
def check_olt_status(customer_id):
   """Check the status of OLT for a specific customer."""
   customer_info = integrated_data.get_customer_info(customer_id)
   if 'OLT_Status' in customer_info:
       return f"The OLT status for customer {customer_id} is: {customer_info['OLT_Status']['Status']}"
   else:
       return f"No OLT status information found for customer {customer_id}"
def get_customer_info(customer_id):
   """Retrieve integrated customer information."""
   customer_info = integrated_data.get_customer_info(customer_id)
   if customer_info:
       return f"Integrated Customer Info for {customer_id}: {customer_info}"
   else:
       return f"No information found for customer {customer_id}"
def create_ticket(customer_id, issue):
   """Create a new support ticket."""
   new_ticket_id = len(integrated_data.ticketing_data) + 1
   print(f"[mock] Creating ticket {new_ticket_id} for customer {customer_id}: {issue}")
   return f"Ticket {new_ticket_id} created successfully for customer {customer_id}."
def get_acs_info(customer_id):
   """Retrieve ACS-specific information for a customer."""
   customer_info = integrated_data.get_customer_info(customer_id)
   if 'ACS' in customer_info:
       return f"ACS Info for customer {customer_id}: {customer_info['ACS']}"
   else:
       return f"No ACS information found for customer {customer_id}"

4.实施代理

SWARM编排层的核心是专门的代理，每个代理都旨在处理系统内的特定任务。在我们的电信示例中，我们定义了几个关键代理：分类代理，它充当初始联系点，并将查询引导给适当的专家；OLT代理，负责处理光线路终端数据；CRM代理，其检索和管理客户信息；ACS代理，处理自动配置服务器数据；以及与知识图集成以解决复杂客户问题的解决方案代理。通过明确定义每个代理的角色和能力，我们创建了一个模块化、高效的系统，可以处理各种客户查询和技术任务。

# Implement agents
olt_agent = Agent(
   name="OLT Agent",
   instructions="Handle OLT-related queries and status checks. Summarise the output 
   in better readble format.",
   functions=[check_olt_status, load_olt_data_dictionary]
)
customer_agent = Agent(
   name="CRM Agent",
   instructions="Handle queries related to the Customer from CRM data. Summarise 
   the output in better readble format.",
   functions=[get_customer_info]
)
acs_agent = Agent(
   name="ACS Agent",
   instructions="Handle queries related to the Auto Configuration Server (ACS) data.
    Summarise the output in better readble format.",
   functions=[get_acs_info]
)
solution_agent = Agent(
   name="Solution Agent",
   instructions="Handle queries related to solving the customer problems, like fixing
    router based on customer data. Summarise the output in better readble format.",
   functions=[knowledgegraph_tool]
)
triage_agent = Agent(
   name="Triage Agent",
   instructions="Determine which agent is best suited to handle the user's request, and 
   transfer the conversation to that agent.",
)

5.定义切换逻辑

SWARM的力量在于它能够无缝地在代理之间转移控制权。我们通过简单的传输函数实现了这一点，如transfer_to_olt（）和transfer_to_crm（）。这些功能支持查询的动态路由，允许代理在遇到超出其专业知识范围的问题时移交任务。例如，CRM代理可以将网络状态查询传递给OLT代理。该机制确保由最合适的专家处理复杂的查询，从而对用户查询做出全面准确的答复。

# Transfer functions
def transfer_to_olt():
   return olt_agent
def transfer_to_customer():
   return customer_agent
def transfer_to_acs():
   return acs_agent
def transfer_to_solution():
   return solution_agent
def transfer_back_to_triage():
   return triage_agent
# Set up transfer functions for each agent
triage_agent.functions = [transfer_to_olt, transfer_to_customer, 
transfer_to_acs, transfer_to_solution]
olt_agent.functions.append(transfer_back_to_triage)
customer_agent.functions.append(transfer_back_to_triage)
acs_agent.functions.append(transfer_back_to_triage)
solution_agent.functions.append(transfer_back_to_triage)

6.让我们执行！！！

run_demo_loop（）将执行一个无限while循环，让代理运行并完成其惊人的工作。

from swarm.repl import run_demo_loop
if __name__ == "__main__":
   run_demo_loop(triage_agent)

结果

此实现完美地说明了编排层的功能。该系统展示了其以下能力：

• 为每个查询识别并激活最合适的代理
• 在后台确定并调用适当的函数或工具
• 从各种数据源检索相关信息
• 在整个对话过程中保持上下文
• 无缝地协助用户进行查询

例如，当被问及客户数据时，系统会激活CRM代理，检索有关客户的详细信息。然后，它会无缝转换为在请求时提供ACS数据，保持之前交互的上下文。当系统正确地将“他”解释为前面提到的客户David时，系统理解上下文的能力得到了进一步的证明。

结论

SWARM框架为构建AI编排层提供了一种强大而灵活的方法。通过将复杂的任务分解为专门的代理并管理它们的交互，您可以创建能够处理各种查询和场景的复杂AI系统。正如我们的电信示例所示，SWARM可以有效地应用于现实世界的用例，为构建智能多代理系统提供了坚实的基础。

👨‍💻 The complete code is available here: