多智能体系统的结构化输出
来源: https://developers.openai.com/cookbook/examples/structured_outputs_multi_agent 抓取时间: 2026-07-21 16:19:02
在本指南中，我们将探索如何使用结构化输出来构建多智能体系统。
结构化输出是一种新功能，它建立在 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
}
]
工具执行
我们需要编写代码逻辑来：
- 处理将用户查询传递给多智能体系统
- 处理多智能体系统的内部工作
- 执行工具调用
为简洁起见，我们将仅定义与用户查询相关的工具逻辑。
# 示例查询
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()
# 定义执行工具的函数
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':
# 模拟数据清理
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':
# 模拟数据转换
transformed_data = {"transformed_data": "sample_transformed_data"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(transformed_data)})
elif tool_name == 'aggregate_data':
# 模拟数据聚合
aggregated_data = {"aggregated_data": "sample_aggregated_data"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(aggregated_data)})
elif tool_name == 'stat_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':
# 模拟相关性分析
correlations = {"correlations": "sample_correlations"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(correlations)})
elif tool_name == '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':
# 模拟条形图创建
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':
# 模拟折线图创建
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':
# 模拟饼图创建
pie_chart = {"pie_chart": "sample_pie_chart"}
messages.append({"role": "tool", "name": tool_name, "content": json.dumps(pie_chart)})
return messages
接下来，我们将为每个子智能体创建工具处理器。 它们有一个独特的提示词和传递给模型的工具集。 然后将输出传递给执行函数，该函数运行工具调用。 我们还将把消息追加到对话历史中。
# 定义处理每个智能体处理的函数
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)
最后，我们创建一个总括性工具来处理用户查询的处理。 此函数接收用户查询，从模型获取响应，并处理将其传递给其他智能体以执行。除此之外，我们还将保持正在进行的对话的状态。
# 处理用户输入和分类的函数
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"}'}]
结论
在本指南中，我们探索了如何利用结构化输出来构建更健壮的多智能体系统。 结构化输出确保工具调用遵循指定的架构。应用程序代码仍必须验证参数语义和业务规则。 这可以应用于更多的用例，我们希望您能从中获得灵感来构建您自己的用例！