01 多智能体协作机制的背景概述
在前述博客中，我们利用LangChain、AutoGen等开发框架构建了一个数据多智能体的平台，并使用了LangChain的Multi-Agents框架。然而，在实施过程中，我们发现现有的框架存在一些局限性，这些局限性影响了系统的整体性能和用户体验。本文将从多智能体协作的重要性、协作机制的难点以及应用方向三个方面进行详细阐述。
多智能体协作的重要性
在现代复杂的信息环境中，单一智能体往往难以应对各种复杂的任务和挑战。因此，多智能体协作机制应运而生。多智能体系统由多个具有自主决策能力的智能体组成，它们通过相互合作与协调，共同完成复杂任务。这种协作机制的重要性和优势主要体现在以下几个方面：
- 提高效率：多智能体协同工作可以显著提升任务执行的速度和效率，特别是在大规模数据处理和分析场景下。
- 增强鲁棒性：单个智能体可能会因为某些原因失效，但多智能体系统的冗余设计可以确保整体系统的稳定运行。
- 适应性强：面对不断变化的需求和环境，多智能体系统能够快速调整策略，灵活应对各种情况。
- 创新潜力：不同智能体之间的交互和学习过程有助于产生新的解决方案和思路，促进技术创新和发展。
协作机制的难点
尽管多智能体协作带来了诸多好处，但在实际应用中也面临着一些挑战和难点：
-
严重依赖模型综合素质：
- 问题：现有的多智能体平台，如LangChain和AutoGen，严重依赖OpenAI的强大模型能力。这些模型虽然在通用任务上表现出色，但在特定领域或专业任务中，其性能可能不尽如人意。
-
与专家系统结合时的问题：
- 问题：在处理非专业领域任务时，多智能体系统表现良好，但当需要与专家系统结合时，会出现一些问题。例如，智能体可能无法准确理解专家系统的输入要求，导致任务执行失败。
-
智能体之间的信息传递问题：
- 问题：智能体之间的信息传递存在不准确或不及时的问题。特别是从用户的问题中抽取专家系统所需的参数时，智能体可能无法准确识别和提取关键信息。
应用方向
多智能体协作机制的应用领域广泛，涵盖了从科学研究到商业实践的各个层面：
-
自动驾驶技术：
- 应用：车辆间的协同驾驶可以帮助减少交通事故发生率，提高道路通行效率。
- 挑战：需要解决多车辆之间的实时通信和协调问题，确保安全和高效的行驶。
-
医疗健康服务：
- 应用：通过集成多种诊断工具和专家意见，为患者提供个性化治疗建议。
- 挑战：需要处理大量医疗数据，并确保数据的安全性和隐私保护。
-
金融风险管理：
- 应用：利用大数据分析预测市场趋势，辅助金融机构做出明智的投资决策。
- 挑战：需要处理复杂的金融数据和模型，确保预测的准确性和可靠性。
-
环境保护监测：
- 应用：部署传感器网络收集环境数据，实时监控污染源并制定有效治理计划。
- 挑战：需要处理大量的实时数据，并确保数据的准确性和及时性。
-
教育科技融合：
- 应用：开发智能化教学平台，支持远程教育和个性化学习体验。
- 挑战：需要结合学生的学习习惯和能力，提供个性化的教学内容和反馈。
总之，多智能体协作机制作为一种新兴的技术手段，在推动社会进步和经济发展方面发挥着越来越重要的作用。未来的研究将继续探索其潜在价值，克服现有障碍，使之成为构建智慧城市的基石之一。通过不断优化和改进，多智能体系统将在更多领域展现出强大的应用潜力。
02 技术框架
在前述的博客中【LLM 构建Data Multi-Agents 赋能数据分析平台的实践之⑤：数据多智能体协作平台的搭建】，我们设计了一个多智能体协作的技术框架，该框架旨在通过多个智能体的合作来解决复杂的问题。本文从协作机制上及解决任务规划、信息传递、参数传递上做了优化。为此，我们设计了多个辅助模块，包括:
任务拆解器 (Task Decomposer): 将用户的问题分解为一系列子任务。
提示词传递器 (Prompt Transmitter): 根据任务描述生成相应的提示词，用于指导后续的智能体操作。
全局参数传递器 (Global Parameter Transmitter Agent): 管理和传递各智能体间共享的参数和状态信息。
任务输出优化器 (Task Result Optimizer): 对最终结果进行优化，以满足用户需求;
任务规划 (Task Planning)
- 任务拆解器 (Task Decomposer):
- 功能: 将用户的问题分解为多个子任务，每个子任务对应一个具体的智能体。
- 优化:
- 动态任务拆解: 根据任务的复杂度和类型，动态调整任务拆解的方式。
- 任务优先级管理: 为不同子任务设置优先级，确保重要任务优先执行。
信息传递 (Information Transmission)
- 提示词传递器 (Prompt Transmitter):
- 功能: 生成并传递提示词，指导智能体执行特定任务。
- 优化:
- 自适应提示词生成: 根据任务的上下文和历史数据，自动生成更精确的提示词。
- 多模态信息传递: 支持文本、图像、音频等多种信息类型的传递。
参数传递 (Parameter Transmission)
- 全局参数传递器 (Global Parameter Transmitter Agent):
- 功能: 管理和传递各智能体间共享的参数和状态信息。
- 优化:
- 分布式参数管理: 采用分布式存储和管理参数，提高系统的可扩展性和容错性。
- 实时参数同步: 确保各智能体之间的参数实时同步，避免信息滞后。
03 流程设计以及Prompt 设计
流程设计
我们设计了一个多智能体协作平台的工作流程。以下是关键步骤和组件：
- 任务输入 (task): 用户或系统提供一个初始任务。
- 任务拆解器 (Task Decomposer): 将原始任务分解成多个子任务（task 1, task 2, …, task N）。
- 全局参数传递器 Agent: 负责在整个过程中传递全局参数。
- 提示词传递器 prompt: 根据需要向各个Agent传递提示词。
- RAG Agent: 处理任务并生成输出。
- 其他Agent (Agent N): 同样处理任务并生成输出。
- 任务结果优化器 Task Result: 对所有任务的结果进行整合和优化。
- 最终输出 Final output: 整合后的最终结果。
在这个过程中，不同的Agent根据各自的能力和功能来完成特定的任务，并通过全局参数传递器和提示词传递器协同工作，确保整个系统的高效运行。
提示词设计
五个主要的设计原则，具体如下：
助手概述：提供关于助手的基本介绍和功能概述。
输出结果形式：明确助手输出结果的具体格式和结构。
Agents描述信息：详细说明与助手相关的各种代理（Agents）的信息。
用户信息：收集和使用用户的个人信息以提供个性化服务。
各步骤输出信息：记录和显示每个操作步骤的输出信息，以便于跟踪和分析。
例如：
messages = [
(
"system",
info
),
("human", """You are a multi-agent collaboration information facilitator.
Based on the user's question, please output a dictionary containing the global variables for each agent involved in the collaboration,
and save it in JSON format.The first step is to review the parameters required by the tools provided by the user.
If the user's question does not involve a particular agent, set the variable parameters for that agent to None.
All user_message information output should ensure that the user input information is stored as much as possible (input:info).
The dictionary should encompass parameters for all tools and look similar to the following structure:
{
'user_message_for_webscrap': '',
'urls_for_webscrap': [""],
'user_message_for_pandas': '',
'OUTPUT_FOLDER': '/content',
'user_message_for_sql': '/content',
'uploaded_file_path_for_sql': '',
'user_message_for_rag': '',
'uploaded_file_path_for_rag': '',
'user_message_for_websearch':''
.....
}
The multi-agent tool library includes:""" + str(tool_info)
),
]
04 代码实践
重构智能体中心，使用DataAgentManager 类将各个智能体（如知识检索、SQL数据分析、Pandas数据分析、网络爬虫、网络搜索等）整合在一起。
##Agents Center
class DataAgentManager:
def __init__(self, config):
self.config = config
# 知识检索功能
def perform_knowledge_retrieval(self):
user_message = self.config['user_message_for_rag']
uploaded_file_path = self.config['uploaded_file_path_for_rag']
......
return output['answer']
def sql_data_analy(self):
user_message = self.config['user_message_for_sql']
file_path = self.config['uploaded_file_path_for_sql']
OUTPUT_FOLDER = self.config['OUTPUT_FOLDER']
....
return chat_result.summary
def pandas_data_analy(self):
user_message = self.config['user_message_for_pandas']
#file_path = self.config['uploaded_file_path_for_pandas']
file_path = '/content'
OUTPUT_FOLDER = self.config['OUTPUT_FOLDER']
.....
return chat_result.summary
def webscrap_data_agent(self):
prompt = self.config['user_message_for_webscrap']
urls = self.config['urls_for_webscrap']
....
return ai_msg.content
def websearch_agent(self):
url = self.config['user_message_for_websearch']
print(url)
多智能体的详细信息设计：智能体概述、智能体预期输出、智能体所需参数
agent_info = {
"agents": [
{
"name": "RAG_agent",
"description": "Retrieval-Augmented-Generation; A RAG too, By retrieving relevant text, obtain knowledge/industry standards/data analysis strategies, which related to user questions, and summarize and generate answers.",
"Agent Related parameters" : "'user_message_for_rag','uploaded_file_path_for_rag'"
},
{
"name": "sql_agent",
"description": "Sql-Code-Write-Execute; A tool to assist users in writing SQL queries and to be able to securely execute those queries in the specified database environment. Function Description: Query building: Construct valid SQL query statements based on the user's natural language requests. Query Execution: Securely execute built SQL queries in the specified database environment. Result display: Returns the query results to the user in an easy-to-understand format. Error handling: Handles any errors that may occur during SQL execution and provides helpful feedback.",
"Agent Related parameters" : "'OUTPUT_FOLDER','user_message_for_sql','uploaded_file_path_for_sql'"
},
{
"name": "csv_agent",
"description": "csv-data-analysis; A tool to assist users in writing python code to analyze CSV data. Solve tasks using your coding and language skills. In the following cases, suggest Python code (in a Python coding block) or shell script (in a sh coding block) for the user to execute. 1. When you need to collect info, use the code to output the info you need, for example, browse or search the web, download/read a file, print the content of a webpage or a file, get the current date/time, check the operating system. After sufficient info is printed and the task is ready to be solved based on your language skill, you can solve the task by yourself. 2. When you need to perform some task with code, use the code to perform the task and output the result. Finish the task smartly. Solve the task step by step if you need to. If a plan is not provided, explain your plan first. Be clear which step uses code, and which step uses your language skill. When using code, you must indicate the script type in the code block. The user cannot provide any other feedback or perform any other action beyond executing the code you suggest. The user can't modify your code. So do not suggest incomplete code which requires users to modify. Don't use a code block if it's not intended to be executed by the user. If you want the user to save the code in a file before executing it, put # filename: <filename> inside the code block as the first line. Don't include multiple code blocks in one response. Do not ask users to copy and paste the result. Instead, use 'print' function for the output when relevant. Check the execution result returned by the user. If the result indicates there is an error, fix the error and output the code again. Suggest the full code instead of partial code or code changes. If the error can't be fixed or if the task is not solved even after the code is executed successfully, analyze the problem, revisit your assumption, collect additional info you need, and think of a different approach to try. When you find an answer, verify the answer carefully. Include verifiable evidence in your response if possible.",
"Agent Related parameters" : "'OUTPUT_FOLDER','user_message_for_pandas','uploaded_file_path_for_pandas'"
},
{
"name": "webscrap_data_agent",
"description": "webscrap_data_agent; You are a web data retrieval and organization assistant. Your main task is to retrieve relevant data from the URLs provided by the user and process it to be structured, then output it in JSON format. The output JSON should include all key information and be easy to understand and use.If the user does not provide a specific URL, this agent will not be able to work.",
"Agent Related parameters" : "'user_message_for_webscrap','urls_for_webscrap'"
},
{
"name": "websearch_agent",
"description": "websearch_agent; You are a web search tool, and your main task is to search through a vast number of web pages to find the most relevant content related to the user's question. The output should include the URL and a brief summary (no more than 50 words) of the web page.",
"Agent Related parameters" : "'user_message_for_websearch'"
},
]
}
辅助模块设计：任务拆解(规划)器、提示词信息传递器、全局参数传递器
任务拆解（规划）器设计：
##任务拆解
def task_Decomposer(prompt,tool_info,BASE_PATH,):
llm = ChatOpenAI(openai_api_base="", openai_api_key="", model="")
messages_cob = [
(
"system",
"""You are a task guidance and disassembly tool for a multi-agent collaborative system,
please disassemble the user's problem according to the existing tools.
The output includes tool selection (subtask description, tool selection, specific steps), task execution steps, and summary.
The final output contains a json file:
{
"tool_selection": [
{
"subtask_description": "...",
"tool_selection": "...",
"specific_steps": "..."
},
{
"subtask_description": "...",
"tool_selection": "...",
"specific_steps": "..."
},
],
"task_execution_steps": [
{
"step": 1,
"action": "....",
'tool_selection': '...',
"expected_output": "..."
},
{
"step": 2,
"action": "...",
'tool_selection': '...',
"expected_output": "..."
},
],
"agents_selection_dict":[
agents = {
'websearch_agent': data_manager.websearch_agent(),
'webscrap_data_agent': data_manager.webscrap_data_agent(),
'csv_agent': data_manager.pandas_data_analy()
}
]
"summary": "..."
}
"""+"\n Tools includes:" + str(tool_info)),
("human", prompt),
]
cob_result_ = llm.invoke(messages_cob)
cob_result = cob_result_.content
json_str = cob_result.replace('```json', '').replace('```', '').strip()
json_data = json.loads(json_str)
json_path = os.path.join(BASE_PATH, 'task_decomposer_output.json')
return json_data
- 任务拆解器将用户的任务拆解成各个智能体能处理的步骤，输出的形式是一个json格式文件，可供后续流程使用。
**提示词传递器：**提示词传递器承载了各个步骤的输出信息，可以流转到后续流程使用
##提示词传递模块
def prompt_info_agent(step,cob_result,step_result):
llm = ChatOpenAI(openai_api_base="", openai_api_key="", model="")
task_info = """You are part of a data multi-agent collaboration platform, and your main task is to output the prompts needed by the next agent based on the execution of each task."""
info = (
task_info +
"\nThe user's overall task and execution steps are as follows: " + str(cob_result) + "; " +
"the output of step: " + str(step) + " is " + str(step_result) + ". " +
"Based on the requirements for the next step's task execution, generate the prompt needed by the next agent."
)
messages = [
(
"system",
info
),
("human", 'Based on the task description and the output results of each step, consolidate and generate the prompt for the next step.Ensure that the outputs from the previous steps are maximally preserved and accurately reflected in the new prompt.'),
]
ai_msg = llm.invoke(messages)
return ai_msg.content
全局参数传递器：根据任务情况、各步骤输出情况迭代生成多智能体协作中心所需的全局变量
##参数传递
def global_config(info,tool_info):
#info:任务输入
#tool_info: 工具或者agents描述信息，包括名称、描述、预期输出、参数
#输出为多智能体协作的全局变量
llm = ChatOpenAI(openai_api_base="", openai_api_key="", model="")
messages = [
(
"system",
info
),
("human", """You are a multi-agent collaboration information facilitator.
Based on the user's question, please output a dictionary containing the global variables for each agent involved in the collaboration,
and save it in JSON format.The first step is to review the parameters required by the tools provided by the user.
If the user's question does not involve a particular agent, set the variable parameters for that agent to None.
All user_message information output should ensure that the user input information is stored as much as possible (input:info).
The dictionary should encompass parameters for all tools and look similar to the following structure:
{
'user_message_for_webscrap': '',
'urls_for_webscrap': [""],
'user_message_for_pandas': '',
'OUTPUT_FOLDER': '/content',
'user_message_for_sql': '/content',
'uploaded_file_path_for_sql': '',
'user_message_for_rag': '',
'uploaded_file_path_for_rag': '',
'user_message_for_websearch':''
.....
}
The multi-agent tool library includes:""" + str(tool_info)
),
]
ai_msg = llm.invoke(messages)
cob_result=ai_msg.content
print(str(cob_result))
global_config = json.loads(cob_result.replace('```json', '').replace('```', '').strip())
return global_config
例如：用户的问题是：info = “从https://www.cnhnb.com/hangqing/changyuo/检索鲳鱼价格信息，转化成json格式，分析各地区鲳鱼价格趋势，包括平均值、最大最小值，绘制一个鲳鱼价格数据看板，输出结果为csv”
全局参数传递器将生成如下全局参数：
```json
{
"user_message_for_webscrap": "检索鲳鱼价格信息，转化成json格式",
"urls_for_webscrap": ["https://www.cnhnb.com/hangqing/changyuo/"],
"user_message_for_pandas": "分析各地区鲳鱼价格趋势，包括平均值、最大最小值，绘制一个鲳鱼价格数据看板，输出结果为csv",
"OUTPUT_FOLDER": "/content",
"user_message_for_sql": null,
"uploaded_file_path_for_sql": null,
"user_message_for_rag": null,
"uploaded_file_path_for_rag": null,
"user_message_for_websearch": null
}
05 测试
info = "从https://www.cnhnb.com/hangqing/changyuo/检索鲳鱼价格信息，转化成json格式，分析各地区鲳鱼价格趋势，包括平均值、最大最小值，绘制一个鲳鱼价格数据看板，输出结果为csv"
#任务拆解
task_config = task_Decomposer(info,agent_info,'/content',)
#agent 参数
global_config_ = global_config(info,agent_info)
data_manager = DataAgentManager(global_config_)
agents = {'websearch_agent': 'data_manager.websearch_agent()',
'webscrap_data_agent': 'data_manager.webscrap_data_agent()',
'csv_agent': 'data_manager.pandas_data_analy()'}
#步骤1输出
out_put = eval(agents[task_config['task_execution_steps'][0]['tool_selection']])
out_put2 = prompt_info_agent(1,task_config,out_put)
global_config_ = global_config(out_put2,agent_info)
data_manager = DataAgentManager(global_config_)
# 步骤2 输出
out_put3 = eval(agents[task_config['task_execution_steps'][1]['tool_selection']])
- 任务拆解出来的结果：
- 从网址获取的鲳鱼数据，结构化为json格式：
{
"metadata": {
"source": "https://www.cnhnb.com/hangqing/changyuo/",
"title": "鲳鱼最新产地行情_鲳鱼产地行情数据_鲳鱼批发价格查询 - 惠农网",
"date": "2024-11-04",
"overview": "惠农网提供实时精准的鲳鱼产地行情查询功能，汇聚最新最全的水果蔬菜、畜牧水产、农副产品等农产品产地行情数据。"
},
"data": {
"prices": [
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "辽宁沈阳市沈北新区",
"price": "5.1元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "辽宁大连市甘井子区",
"price": "12.75元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "福建福州市马尾区",
"price": "5.74元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "福建漳州市东山县",
"price": "15.3元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "山东济南市历城区",
"price": "4.68元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "山东威海市荣成市",
"price": "4.89元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "鲳鱼",
"location": "山东日照市岚山区",
"price": "5.1元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "辽宁大连市甘井子区",
"price": "4.68元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "江苏南通市启东市",
"price": "10.2元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "江苏连云港市赣榆区",
"price": "13.16元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "浙江台州市临海市",
"price": "14.29元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "山东青岛市崂山区",
"price": "10.52元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "山东济宁市任城区",
"price": "7.54元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "山东威海市荣成市",
"price": "5.1元/斤",
"trend": "降"
},
{
"date": "2024-11-04",
"product": "银鲳鱼",
"location": "山东日照市东港区",
"price": "11.59元/斤",
"trend": "降"
}
]
}
}
- 多智能体间的信息传递：提示词传递器生成下一步骤的提示信息
- 下一步骤相关结果：pandas_agent分析从网页获取的鲳鱼价格数据
- 生成的代码
# filename: pomfret_price_analysis.py
import pandas as pd
import streamlit as st
def load_data():
# Load the JSON data into a DataFrame
df = pd.read_json('C:/Users/liuli/Desktop/output.json')
return df
def calculate_price_index(df):
# Filter out rows where 'data' is not a list
df = df[df['data'].apply(lambda x: isinstance(x, list))]
# Extract the prices of Pomfret
pomfret_data = df['data'].apply(lambda x: [item for item in x if item['product'] == '鲳鱼'])
# Flatten the list of dictionaries
pomfret_list = [item for sublist in pomfret_data for item in sublist]
# Convert the extracted data into a DataFrame
pomfret_df = pd.DataFrame(pomfret_list)
# Clean and convert the price strings to numeric values
pomfret_df['price'] = pomfret_df['price'].str.replace('元/斤', '').astype(float)
# Calculate the average, maximum, and minimum prices
pomfret_df['average_price'] = pomfret_df.groupby('location')['price'].transform('mean')
pomfret_df['max_price'] = pomfret_df.groupby('location')['price'].transform('max')
pomfret_df['min_price'] = pomfret_df.groupby('location')['price'].transform('min')
return pomfret_df
def create_dashboard(df):
st.title('Pomfret Price Trend Dashboard')
# Display the average, maximum, and minimum prices
st.write('### Average Price')
st.line_chart(df.set_index('location')['average_price'])
st.write('### Maximum Price')
st.line_chart(df.set_index('location')['max_price'])
st.write('### Minimum Price')
st.line_chart(df.set_index('location')['min_price'])
def main():
df = load_data()
df = calculate_price_index(df)
create_dashboard(df)
if __name__ == '__main__':
main()
- 生成的数据看板
06 结果讨论
-
在产业数字化转型的过程中，多智能体协作机制与流程扮演了至关重要的角色。为了充分释放大型语言模型（LLM）在这一过程中的潜力，关键在于如何巧妙地利用其强大的任务分解、意图解析、编程能力和数据处理技巧，来实现与专家系统及特定领域模型的有效融合与协同工作。这种整合不仅能够增强系统的整体功能，还能提高应对复杂业务场景的灵活性和适应性。
-
为了解决多智能体间的信息流通问题，本文提出了一种基于提示词的通信机制，即构建一个“提示词传递器”，用以在不同智能体之间高效地传输子任务描述、执行结果等信息。此外，我们还设计了一个“全局参数传递器”，专门负责在专家系统、特定模型与智能体之间共享任务执行过程中产生的关键数据，例如URL链接、特定的数据条目、文件资料及提示指令等。这两种机制的引入，极大地促进了多智能体系统内部的沟通与合作效率。
-
实践证明，在多智能体协作框架下，大型语言模型的表现直接关系到整个系统运行的效果。具体来说，能否将复杂的任务合理地分解成一系列简单且具体的子任务，是否能准确无误地生成和传递全局参数，以及所编写的代码是否既正确又具有良好的执行性能，这些都成为了衡量大模型能力高低的重要指标。因此，提升大模型的上述各方面技能，对于确保多智能体协作的成功至关重要。
-
随着多智能体协作理论与技术的不断进步，未来很可能会出现针对不同应用场景而定制的专业型大模型。比如，专门为项目管理设计的大模型，能够高效地完成任务规划与调度；或者是专注于参数优化的大模型，能够在短时间内为用户提供最佳配置建议。这些专业化的解决方案将进一步推动产业数字化进程，帮助企业在激烈的市场竞争中获得更大的优势。
![[Linux] 进程控制之创建和终止](https://i-blog.csdnimg.cn/direct/5ce10a4ce24f4272815c42035527d679.png)
