在本 cookbook 中，我们将探索如何使用结构化输出来构建多智能体系统。
结构化输出是一项新功能，它建立在 JSON 模式和函数调用的基础上，以在模型输出中强制执行严格的模式。
通过使用新的参数 strict: true，我们能够保证响应遵守提供的模式。
为了演示此功能的强大之处，我们将使用它来构建一个多智能体系统。
当使用函数调用时，如果函数（或工具）的数量增加，性能可能会受到影响。
为了缓解这种情况，我们可以将工具进行逻辑分组，并让专门的“智能体”能够解决特定的任务或子任务，这将提高整体系统性能。
from openai import OpenAI
from IPython.display import Image
import json
import pandas as pd
import matplotlib.pyplot as plt
from io import StringIO
import numpy as np
client = OpenAI()MODEL = "gpt-4o-2024-08-06"我们将要解决的用例是数据分析任务。
让我们首先设置我们的 4 智能体系统
我们将首先定义每个智能体的系统提示。
triaging_system_prompt = """You are a Triaging Agent. Your role is to assess the user's query and route it to the relevant agents. The agents available are:
- Data Processing Agent: Cleans, transforms, and aggregates data.
- Analysis Agent: Performs statistical, correlation, and regression analysis.
- Visualization Agent: Creates bar charts, line charts, and pie charts.
Use the send_query_to_agents tool to forward the user's query to the relevant agents. Also, use the speak_to_user tool to get more information from the user if needed."""
processing_system_prompt = """You are a Data Processing Agent. Your role is to clean, transform, and aggregate data using the following tools:
- clean_data
- transform_data
- aggregate_data"""
analysis_system_prompt = """You are an Analysis Agent. Your role is to perform statistical, correlation, and regression analysis using the following tools:
- stat_analysis
- correlation_analysis
- regression_analysis"""
visualization_system_prompt = """You are a Visualization Agent. Your role is to create bar charts, line charts, and pie charts using the following tools:
- create_bar_chart
- create_line_chart
- create_pie_chart"""然后，我们将定义每个智能体的工具。
除了分诊智能体外，每个智能体都将配备特定于其角色的工具
triage_tools = [
{
"type": "function",
"function": {
"name": "send_query_to_agents",
"description": "Sends the user query to relevant agents based on their capabilities.",
"parameters": {
"type": "object",
"properties": {
"agents": {
"type": "array",
"items": {"type": "string"},
"description": "An array of agent names to send the query to."
},
"query": {
"type": "string",
"description": "The user query to send."
}
},
"required": ["agents", "query"]
}
},
"strict": True
}
]
preprocess_tools = [
{
"type": "function",
"function": {
"name": "clean_data",
"description": "Cleans the provided data by removing duplicates and handling missing values.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The dataset to clean. Should be in a suitable format such as JSON or CSV."
}
},
"required": ["data"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "transform_data",
"description": "Transforms data based on specified rules.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The data to transform. Should be in a suitable format such as JSON or CSV."
},
"rules": {
"type": "string",
"description": "Transformation rules to apply, specified in a structured format."
}
},
"required": ["data", "rules"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "aggregate_data",
"description": "Aggregates data by specified columns and operations.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The data to aggregate. Should be in a suitable format such as JSON or CSV."
},
"group_by": {
"type": "array",
"items": {"type": "string"},
"description": "Columns to group by."
},
"operations": {
"type": "string",
"description": "Aggregation operations to perform, specified in a structured format."
}
},
"required": ["data", "group_by", "operations"],
"additionalProperties": False
}
},
"strict": True
}
]
analysis_tools = [
{
"type": "function",
"function": {
"name": "stat_analysis",
"description": "Performs statistical analysis on the given dataset.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The dataset to analyze. Should be in a suitable format such as JSON or CSV."
}
},
"required": ["data"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "correlation_analysis",
"description": "Calculates correlation coefficients between variables in the dataset.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The dataset to analyze. Should be in a suitable format such as JSON or CSV."
},
"variables": {
"type": "array",
"items": {"type": "string"},
"description": "List of variables to calculate correlations for."
}
},
"required": ["data", "variables"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "regression_analysis",
"description": "Performs regression analysis on the dataset.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The dataset to analyze. Should be in a suitable format such as JSON or CSV."
},
"dependent_var": {
"type": "string",
"description": "The dependent variable for regression."
},
"independent_vars": {
"type": "array",
"items": {"type": "string"},
"description": "List of independent variables."
}
},
"required": ["data", "dependent_var", "independent_vars"],
"additionalProperties": False
}
},
"strict": True
}
]
visualization_tools = [
{
"type": "function",
"function": {
"name": "create_bar_chart",
"description": "Creates a bar chart from the provided data.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The data for the bar chart. Should be in a suitable format such as JSON or CSV."
},
"x": {
"type": "string",
"description": "Column for the x-axis."
},
"y": {
"type": "string",
"description": "Column for the y-axis."
}
},
"required": ["data", "x", "y"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "create_line_chart",
"description": "Creates a line chart from the provided data.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The data for the line chart. Should be in a suitable format such as JSON or CSV."
},
"x": {
"type": "string",
"description": "Column for the x-axis."
},
"y": {
"type": "string",
"description": "Column for the y-axis."
}
},
"required": ["data", "x", "y"],
"additionalProperties": False
}
},
"strict": True
},
{
"type": "function",
"function": {
"name": "create_pie_chart",
"description": "Creates a pie chart from the provided data.",
"parameters": {
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "The data for the pie chart. Should be in a suitable format such as JSON or CSV."
},
"labels": {
"type": "string",
"description": "Column for the labels."
},
"values": {
"type": "string",
"description": "Column for the values."
}
},
"required": ["data", "labels", "values"],
"additionalProperties": False
}
},
"strict": True
}
]我们需要编写代码逻辑来
为了简洁起见，我们将仅定义与用户查询相关的工具的逻辑。
# Example query
user_query = """
Below is some data. I want you to first remove the duplicates then analyze the statistics of the data as well as plot a line chart.
house_size (m3), house_price ($)
90, 100
80, 90
100, 120
90, 100
"""
从用户查询中，我们可以推断出我们需要调用的工具是 clean_data、start_analysis 和 use_line_chart。
我们将首先定义执行函数，该函数运行工具调用。
这会将工具调用映射到相应的函数。然后，它将函数的输出附加到对话历史记录中。
def clean_data(data):
data_io = StringIO(data)
df = pd.read_csv(data_io, sep=",")
df_deduplicated = df.drop_duplicates()
return df_deduplicated
def stat_analysis(data):
data_io = StringIO(data)
df = pd.read_csv(data_io, sep=",")
return df.describe()
def plot_line_chart(data):
data_io = StringIO(data)
df = pd.read_csv(data_io, sep=",")
x = df.iloc[:, 0]
y = df.iloc[:, 1]
coefficients = np.polyfit(x, y, 1)
polynomial = np.poly1d(coefficients)
y_fit = polynomial(x)
plt.figure(figsize=(10, 6))
plt.plot(x, y, 'o', label='Data Points')
plt.plot(x, y_fit, '-', label='Best Fit Line')
plt.title('Line Chart with Best Fit Line')
plt.xlabel(df.columns[0])
plt.ylabel(df.columns[1])
plt.legend()
plt.grid(True)
plt.show()
# Define the function to execute the tools
def execute_tool(tool_calls, messages):
for tool_call in tool_calls:
tool_name = tool_call.function.name
tool_arguments = json.loads(tool_call.function.arguments)
if tool_name == 'clean_data':
# Simulate data cleaning
cleaned_df = clean_data(tool_arguments['data'])
cleaned_data = {"cleaned_data": cleaned_df.to_dict()}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(cleaned_data)})
print('Cleaned data: ', cleaned_df)
elif tool_name == 'transform_data':
# Simulate data transformation
transformed_data = {"transformed_data": "sample_transformed_data"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(transformed_data)})
elif tool_name == 'aggregate_data':
# Simulate data aggregation
aggregated_data = {"aggregated_data": "sample_aggregated_data"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(aggregated_data)})
elif tool_name == 'stat_analysis':
# Simulate statistical analysis
stats_df = stat_analysis(tool_arguments['data'])
stats = {"stats": stats_df.to_dict()}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(stats)})
print('Statistical Analysis: ', stats_df)
elif tool_name == 'correlation_analysis':
# Simulate correlation analysis
correlations = {"correlations": "sample_correlations"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(correlations)})
elif tool_name == 'regression_analysis':
# Simulate regression analysis
regression_results = {"regression_results": "sample_regression_results"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(regression_results)})
elif tool_name == 'create_bar_chart':
# Simulate bar chart creation
bar_chart = {"bar_chart": "sample_bar_chart"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(bar_chart)})
elif tool_name == 'create_line_chart':
# Simulate line chart creation
line_chart = {"line_chart": "sample_line_chart"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(line_chart)})
plot_line_chart(tool_arguments['data'])
elif tool_name == 'create_pie_chart':
# Simulate pie chart creation
pie_chart = {"pie_chart": "sample_pie_chart"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(pie_chart)})
return messages接下来，我们将为每个子智能体创建工具处理程序。
这些程序具有传递给模型的唯一提示和工具集。
然后，输出将传递给执行函数，该函数运行工具调用。
我们还将消息附加到对话历史记录中。
# Define the functions to handle each agent's processing
def handle_data_processing_agent(query, conversation_messages):
messages = [{"role": "system", "content": processing_system_prompt}]
messages.append({"role": "user", "content": query})
response = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0,
tools=preprocess_tools,
)
conversation_messages.append([tool_call.function for tool_call in response.choices[0].message.tool_calls])
execute_tool(response.choices[0].message.tool_calls, conversation_messages)
def handle_analysis_agent(query, conversation_messages):
messages = [{"role": "system", "content": analysis_system_prompt}]
messages.append({"role": "user", "content": query})
response = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0,
tools=analysis_tools,
)
conversation_messages.append([tool_call.function for tool_call in response.choices[0].message.tool_calls])
execute_tool(response.choices[0].message.tool_calls, conversation_messages)
def handle_visualization_agent(query, conversation_messages):
messages = [{"role": "system", "content": visualization_system_prompt}]
messages.append({"role": "user", "content": query})
response = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0,
tools=visualization_tools,
)
conversation_messages.append([tool_call.function for tool_call in response.choices[0].message.tool_calls])
execute_tool(response.choices[0].message.tool_calls, conversation_messages)
最后，我们创建用于处理用户查询的总体工具。
此函数接受用户查询，从模型获取响应，并处理将其传递给其他智能体以执行。除此之外，我们将保留正在进行的对话的状态。
# Function to handle user input and triaging
def handle_user_message(user_query, conversation_messages=[]):
user_message = {"role": "user", "content": user_query}
conversation_messages.append(user_message)
messages = [{"role": "system", "content": triaging_system_prompt}]
messages.extend(conversation_messages)
response = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0,
tools=triage_tools,
)
conversation_messages.append([tool_call.function for tool_call in response.choices[0].message.tool_calls])
for tool_call in response.choices[0].message.tool_calls:
if tool_call.function.name == 'send_query_to_agents':
agents = json.loads(tool_call.function.arguments)['agents']
query = json.loads(tool_call.function.arguments)['query']
for agent in agents:
if agent == "Data Processing Agent":
handle_data_processing_agent(query, conversation_messages)
elif agent == "Analysis Agent":
handle_analysis_agent(query, conversation_messages)
elif agent == "Visualization Agent":
handle_visualization_agent(query, conversation_messages)
return conversation_messages最后，我们在用户查询上运行总体的 handle_user_message 函数，并查看输出。
handle_user_message(user_query)Cleaned data: house_size (m3) house_price ($) 0 90 100 1 80 90 2 100 120 Statistical Analysis: house_size house_price count 4.000000 4.000000 mean 90.000000 102.500000 std 8.164966 12.583057 min 80.000000 90.000000 25% 87.500000 97.500000 50% 90.000000 100.000000 75% 92.500000 105.000000 max 100.000000 120.000000
[{'role': 'user',
'content': '\nBelow is some data. I want you to first remove the duplicates then analyze the statistics of the data as well as plot a line chart.\n\nhouse_size (m3), house_price ($)\n90, 100\n80, 90\n100, 120\n90, 100\n'},
[Function(arguments='{"agents": ["Data Processing Agent"], "query": "Remove duplicates from the data: house_size (m3), house_price ($)\\n90, 100\\n80, 90\\n100, 120\\n90, 100"}', name='send_query_to_agents'),
Function(arguments='{"agents": ["Analysis Agent"], "query": "Analyze the statistics of the data: house_size (m3), house_price ($)\\n90, 100\\n80, 90\\n100, 120\\n90, 100"}', name='send_query_to_agents'),
Function(arguments='{"agents": ["Visualization Agent"], "query": "Plot a line chart for the data: house_size (m3), house_price ($)\\n90, 100\\n80, 90\\n100, 120\\n90, 100"}', name='send_query_to_agents')],
[Function(arguments='{"data":"house_size (m3), house_price ($)\\n90, 100\\n80, 90\\n100, 120\\n90, 100"}', name='clean_data')],
{'role': 'tool',
'name': 'clean_data',
'content': '{"cleaned_data": {"house_size (m3)": {"0": 90, "1": 80, "2": 100}, " house_price ($)": {"0": 100, "1": 90, "2": 120}}}'},
[Function(arguments='{"data":"house_size,house_price\\n90,100\\n80,90\\n100,120\\n90,100"}', name='stat_analysis')],
{'role': 'tool',
'name': 'stat_analysis',
'content': '{"stats": {"house_size": {"count": 4.0, "mean": 90.0, "std": 8.16496580927726, "min": 80.0, "25%": 87.5, "50%": 90.0, "75%": 92.5, "max": 100.0}, "house_price": {"count": 4.0, "mean": 102.5, "std": 12.583057392117917, "min": 90.0, "25%": 97.5, "50%": 100.0, "75%": 105.0, "max": 120.0}}}'},
[Function(arguments='{"data":"house_size,house_price\\n90,100\\n80,90\\n100,120\\n90,100","x":"house_size","y":"house_price"}', name='create_line_chart')],
{'role': 'tool',
'name': 'create_line_chart',
'content': '{"line_chart": "sample_line_chart"}'}]在本 cookbook 中，我们探索了如何利用结构化输出来构建更强大的多智能体系统。
使用这项新功能可以确保工具调用遵循指定的模式，并避免必须处理边缘情况或验证您这边的参数。
这可以应用于更多用例，我们希望您可以从中获得灵感来构建自己的用例！