（多）智能体监督 (Agent Supervision)
概述
智能体监督是LangChain中的一种多智能体协作模式，它引入了一个"监督者"角色来协调和指导其他智能体的工作。在这种模式下，监督智能体负责将复杂任务分解为子任务，分配给适当的专业智能体处理，监控执行过程，处理失败的情况，以及整合各个智能体的输出结果。这种方法特别适合需要多种专业知识协同工作的复杂任务。
智能体监督模式解决了单个智能体在处理复杂、跨领域任务时可能面临的局限性，通过引入明确的层次结构和专业分工，提高了整个系统的性能和可靠性。
工作原理
智能体监督系统的工作流程通常包括以下几个阶段：
任务分析：监督智能体分析用户需求并确定需要的专业智能体类型
任务分解：将主任务分解为更小、更专业的子任务
智能体选择：为每个子任务选择或创建适当的专业智能体
任务分配：将子任务分配给相应的专业智能体
执行监督：监控专业智能体的执行过程，必要时提供指导
结果整合：收集各个智能体的输出并整合为最终结果
错误处理：识别并处理执行过程中的错误或失败
优势
专业分工：允许不同智能体专注于自己擅长的任务领域
复杂问题解决：能够处理需要多种专业知识的复杂问题
更好的错误恢复：监督者可以检测并纠正专业智能体的错误
资源优化：通过精确分配任务减少不必要的计算资源浪费
可扩展性：易于添加新的专业智能体以扩展系统功能
更强的可控性：通过监督者提供对整个系统的中央控制
基本用法示例
以下是创建和使用基本智能体监督系统的完整示例：
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import DuckDuckGoSearchRun, WikipediaQueryRun, Tool
from langchain_community.utilities import WikipediaAPIWrapper
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import SystemMessage
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 创建专业智能体的工具
search_tool = DuckDuckGoSearchRun(name="搜索")
wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(), name="维基百科")
# 创建研究专家智能体
research_tools = [search_tool, wiki_tool]
research_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个研究专家，擅长收集和整理信息。使用提供的工具来查找用户请求的信息。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
research_agent = create_openai_functions_agent(llm, research_tools, research_prompt)
research_executor = AgentExecutor(agent=research_agent, tools=research_tools, verbose=True)
# 创建写作专家智能体
writer_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个写作专家，擅长将信息组织成清晰、连贯的文本。根据提供的信息创建高质量的内容。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
writer_agent = create_openai_functions_agent(llm, [], writer_prompt)
writer_executor = AgentExecutor(agent=writer_agent, tools=[], verbose=True)
# 创建监督智能体的工具（包含专业智能体）
def research_expert(query: str) -> str:
"""使用研究专家智能体查找信息"""
return research_executor.invoke({"input": query})["output"]
def writing_expert(content: str) -> str:
"""使用写作专家智能体创建或编辑内容"""
return writer_executor.invoke({"input": content})["output"]
supervisor_tools = [
Tool(name="研究专家", func=research_expert, description="当你需要查找信息时使用"),
Tool(name="写作专家", func=writing_expert, description="当你需要创建或编辑内容时使用")
]
# 创建监督智能体
supervisor_prompt = ChatPromptTemplate.from_messages([
("system", """你是一个任务监督者，负责协调专家智能体完成复杂任务。
你可以使用以下专家：
1. 研究专家 - 擅长查找和收集信息
2. 写作专家 - 擅长创建高质量内容
你的工作流程：
1. 分析用户的请求
2. 确定需要哪些专家以及执行顺序
3. 逐步调用专家，一次只调用一个
4. 每次提供专家明确具体的指令
5. 基于之前专家的结果指导下一位专家
6. 返回最终的综合结果"""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
supervisor_agent = create_openai_functions_agent(llm, supervisor_tools, supervisor_prompt)
supervisor_executor = AgentExecutor(agent=supervisor_agent, tools=supervisor_tools, verbose=True)
# 执行监督智能体
result = supervisor_executor.invoke(
{"input": "请研究量子计算的基本原理并创建一篇500字的科普文章"}
)
print("最终结果:\n", result["output"])
高级用法
动态智能体创建与分配
智能体监督系统的高级用法之一是动态创建和分配专业智能体。这种方法允许系统根据任务需求自动确定所需的专业知识类型，并创建相应的智能体。
from typing import Dict, List, Any
class AgentSupervisor:
def __init__(self, llm):
self.llm = llm
self.specialized_agents = {}
self.tools = []
self._initialize_system()
def _initialize_system(self):
# 创建监督决策工具
self.tools.append(
Tool(
name="task_analyzer",
func=self._analyze_task,
description="分析任务并决定需要哪些专业智能体"
)
)
def _analyze_task(self, task: str) -> str:
"""分析任务并返回所需专业智能体列表"""
analysis_prompt = f"""分析以下任务，确定完成任务所需的专业智能体类型。
可用的专业类型包括：研究、写作、数据分析、编程、创意生成等。
对于每个建议的专业智能体，说明它的角色和在任务中的职责。
任务：{task}
"""
response = self.llm.predict(analysis_prompt)
return response
def create_specialized_agent(self, agent_type: str, description: str) -> AgentExecutor:
"""创建特定类型的专业智能体"""
# 根据智能体类型选择适当的工具
agent_tools = self._select_tools_for_agent(agent_type)
# 创建智能体提示
agent_prompt = ChatPromptTemplate.from_messages([
("system", f"你是一个专业的{agent_type}智能体。{description}"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# 创建智能体
agent = create_openai_functions_agent(self.llm, agent_tools, agent_prompt)
executor = AgentExecutor(agent=agent, tools=agent_tools, verbose=True)
# 注册智能体
self.specialized_agents[agent_type] = executor
# 创建调用该智能体的工具
self.tools.append(
Tool(
name=f"{agent_type}_expert",
func=lambda input, agent_type=agent_type: self.specialized_agents[agent_type].invoke({"input": input})["output"],
description=f"使用{agent_type}专家来{description}"
)
)
return executor
def _select_tools_for_agent(self, agent_type: str) -> List[Tool]:
"""为不同类型的智能体选择合适的工具"""
if agent_type.lower() == "研究":
return [search_tool, wiki_tool]
elif agent_type.lower() == "数据分析":
return [Tool(name="python_calculator", func=lambda x: eval(x), description="执行Python计算")]
else:
return [] # 某些智能体可能不需要工具
def create_supervisor_agent(self) -> AgentExecutor:
"""创建监督智能体"""
supervisor_prompt = ChatPromptTemplate.from_messages([
("system", """你是一个任务监督者，负责协调各个专业智能体完成复杂任务。
分析任务，将其分解为子任务，并分配给相应的专业智能体。
监督他们的工作，必要时提供额外指导，并整合所有结果。
可用的专业智能体有：
{tools_description}"""),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
tools_desc = "\n".join([f"- {tool.name}: {tool.description}" for tool in self.tools])
supervisor_agent = create_openai_functions_agent(
self.llm,
self.tools,
supervisor_prompt.partial(tools_description=tools_desc)
)
return AgentExecutor(agent=supervisor_agent, tools=self.tools, verbose=True)
监督智能体与错误处理
高级监督系统的重要特性是强大的错误处理机制，能够检测专业智能体的失败并采取适当的恢复措施。
from typing import Optional, Dict
from langchain.callbacks.manager import CallbackManagerForToolRun
class SupervisorWithErrorHandling:
def __init__(self, llm):
self.llm = llm
self.specialist_agents = {}
self.max_retry = 2
self._setup_agents()
def _setup_agents(self):
# 设置专业智能体
# 此处简化，实际应用中可能有更多专业智能体
self.specialist_agents["研究"] = self._create_research_agent()
self.specialist_agents["写作"] = self._create_writing_agent()
def _create_research_agent(self):
# 创建研究智能体
research_tools = [search_tool, wiki_tool]
research_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个研究专家，擅长收集精确的信息。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
research_agent = create_openai_functions_agent(self.llm, research_tools, research_prompt)
return AgentExecutor(agent=research_agent, tools=research_tools, verbose=True)
def _create_writing_agent(self):
# 创建写作智能体
writer_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个写作专家，擅长创建清晰、引人入胜的内容。"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
writer_agent = create_openai_functions_agent(self.llm, [], writer_prompt)
return AgentExecutor(agent=writer_agent, tools=[], verbose=True)
def delegate_with_error_handling(self, agent_type: str, task: str) -> str:
"""委派任务给专业智能体并处理潜在错误"""
if agent_type not in self.specialist_agents:
return f"错误：未找到'{agent_type}'类型的专业智能体"
retries = 0
while retries <= self.max_retry:
try:
result = self.specialist_agents[agent_type].invoke({"input": task})
output = result.get("output", "")
# 验证输出质量
validation = self._validate_output(agent_type, task, output)
if validation["success"]:
return output
else:
# 根据验证反馈修改任务
task = f"{task}\n\n前一次尝试的问题：{validation['feedback']}。请改进你的回答。"
retries += 1
except Exception as e:
task = f"{task}\n\n前一次尝试失败，错误：{str(e)}。请重新尝试。"
retries += 1
# 如果所有重试都失败
return f"警告：'{agent_type}'智能体在{self.max_retry}次尝试后仍无法完成任务。最后的输出：{output}"
def _validate_output(self, agent_type: str, task: str, output: str) -> Dict[str, any]:
"""验证专业智能体的输出质量"""
validation_prompt = f"""评估以下输出是否成功完成了任务。
智能体类型：{agent_type}
任务：{task}
输出：
{output}
请进行以下评估：
1. 输出是否直接回应了任务需求？
2. 内容是否准确、相关？
3. 是否有明显的错误或遗漏？
以JSON格式返回结果：{{
"success": true或false,
"score": 0到10的评分,
"feedback": "具体的改进建议（如果需要）"
}}
"""
try:
response = self.llm.predict(validation_prompt)
import json
validation = json.loads(response)
return validation
except Exception:
# 如果解析失败，假设验证通过
return {"success": True, "score": 7, "feedback": "无法正确解析验证结果"}
实际应用场景
1. 定制内容创建管道
智能体监督系统可以用于构建复杂的内容创建管道，自动执行从研究到最终内容发布的整个过程。
from langchain.tools import tool
from typing import Dict, List
# 创建内容创建专家工具
@tool
def topic_researcher(topic: str) -> str:
"""研究特定主题并返回关键信息"""
research_agent = specialist_agents["研究"]
return research_agent.invoke({
"input": f"搜集关于'{topic}'的最新信息，包括主要观点、统计数据和重要发展。"
})["output"]
@tool
def content_outliner(topic: str, research: str) -> str:
"""基于研究创建内容大纲"""
outline_agent = specialist_agents["大纲"]
return outline_agent.invoke({
"input": f"根据以下研究为'{topic}'创建详细的内容大纲：\n\n{research}"
})["output"]
@tool
def content_writer(topic: str, outline: str, style: str) -> str:
"""根据大纲撰写内容"""
writer_agent = specialist_agents["写作"]
return writer_agent.invoke({
"input": f"根据以下大纲，以'{style}'风格撰写关于'{topic}'的文章：\n\n{outline}"
})["output"]
@tool
def content_editor(draft: str, requirements: str) -> str:
"""编辑和改进内容"""
editor_agent = specialist_agents["编辑"]
return editor_agent.invoke({
"input": f"根据以下要求编辑和改进这篇文章：\n\n要求：{requirements}\n\n草稿：\n{draft}"
})["output"]
2. 跨领域问题解决系统
智能体监督系统特别适合解决需要多个领域专业知识的复杂问题，例如企业决策支持系统。
from langchain.tools import tool
# 创建各领域专家工具
@tool
def data_analyst(data_question: str) -> str:
"""分析数据并回答相关问题"""
# 实际应用中会连接到专门的数据分析智能体
return f"对于'{data_question}'的数据分析结果：[数据分析内容]"
@tool
def legal_expert(legal_question: str) -> str:
"""回答法律相关问题"""
# 实际应用中会连接到法律专家智能体
return f"对于'{legal_question}'的法律见解：[法律分析内容]"
@tool
def technical_expert(technical_question: str) -> str:
"""回答技术相关问题"""
# 实际应用中会连接到技术专家智能体
return f"对于'{technical_question}'的技术解答：[技术内容]"
@tool
def business_strategist(business_question: str) -> str:
"""提供商业战略建议"""
# 实际应用中会连接到商业战略智能体
return f"对于'{business_question}'的商业建议：[战略内容]"
最佳实践
明确职责划分：为每个智能体定义清晰的角色和职责范围。
# 在系统提示中明确职责 system_message = """你是[特定角色]智能体，专门负责[具体职责]。 你应该专注于[主要任务]，不要尝试执行超出你职责的任务。 如果遇到超出范围的问题，请表明这超出了你的专业领域。"""渐进式任务分解：将复杂任务分解为连贯的、有逻辑顺序的子任务。
上下文管理：确保专业智能体获得足够的任务上下文。
def delegate_task(agent_type: str, task: str, context: str = "") -> str: """将任务委派给专业智能体，同时提供上下文""" full_input = f"任务：{task}\n\n上下文：{context}" if context else task return specialist_agents[agent_type].invoke({"input": full_input})["output"]错误处理机制：实现强大的错误检测和恢复策略。
智能体性能监控：持续评估每个智能体的性能并进行必要调整。
def evaluate_agent_performance(agent_type: str, task: str, output: str) -> Dict: """评估智能体的输出质量""" evaluation_prompt = f"""评估以下{agent_type}智能体的输出： 任务：{task} 输出：{output} 评分标准： - 相关性（1-10） - 准确性（1-10） - 完整性（1-10） - 总体质量（1-10） 同时提供具体的改进建议。 """ evaluation = llm.predict(evaluation_prompt) # 解析评估结果 # ... return parsed_evaluation智能任务分配：根据智能体的专长和历史性能分配任务。
结果验证：在接受最终结果前验证其质量和完整性。
反馈循环：实现监督智能体与专业智能体之间的反馈机制。
与其他多智能体模式的对比
常见问题与解决方案
智能体职责重叠
解决方案：明确定义每个智能体的边界和专长领域
实现冲突解决机制，由监督智能体裁决
子任务依赖管理
解决方案：实现任务依赖跟踪系统
使用有向无环图表示任务依赖关系
资源分配不均
解决方案：实现智能资源分配算法
动态调整专业智能体的计算资源
结果不一致
解决方案：加强验证机制
实现结果一致性检查
智能体能力不足
解决方案：实现能力评估机制
提供智能体升级或替换机制
总结
智能体监督是一种强大的多智能体协作模式，特别适合处理需要多种专业知识协作的复杂任务。通过引入监督智能体作为中央协调者，它解决了传统单智能体系统在处理跨领域复杂问题时的局限性，同时提供了更好的错误恢复和任务管理能力。
这种方法的主要优势在于明确的职责分工、结构化的任务分解和协调，以及强大的错误处理机制。通过适当的实现，智能体监督系统可以显著提高复杂任务的处理效率和结果质量。
随着人工智能技术的发展，智能体监督模式将在内容创建、跨领域问题解决、研究与开发等众多领域发挥越来越重要的作用，成为构建复杂智能系统的关键方法之一。
- 感谢你赐予我前进的力量
-
微信 
支付宝
赞赏者名单
因为你们的支持让我意识到写文章的价值🙏
本文是原创文章，采用 CC BY-NC-ND 4.0 协议，完整转载请注明来自 数据·开英
阅读建议
评论
匿名评论
隐私政策
你无需删除空行，直接评论以获取最佳展示效果
