（多）智能体团队 (Agent Team)
概述
智能体团队是一种多智能体协作模式，它将多个智能体组织成一个团队，共同解决复杂任务。与智能体监督模式不同，智能体团队中的成员通常具有平等的地位，通过相互交流和协作来达成目标。这种模式特别适合需要多种观点或多领域专业知识综合考量的问题。
在智能体团队中，每个成员都有特定的角色和专长，但没有明确的层级结构。团队成员可以自由交流，提出建议，并共同决策。这种模式模拟了人类团队协作的方式，强调信息共享和集体智慧。
工作原理
智能体团队的工作流程通常包括以下几个阶段：
团队组建：根据任务需求选择具有互补技能的智能体
任务分析：团队成员共同分析问题和要求
角色分配：根据专长为每个成员分配特定职责
交流与协作：成员之间相互交流信息和见解
共识形成：成员共同讨论并达成解决方案的共识
结果整合：将各成员的贡献整合为最终输出
优势
多角度思考：不同智能体提供多种视角和思考方式
平衡决策：通过集体智慧减少个体偏见
专业互补：结合不同领域的专业知识
创意激发：通过成员间的交流产生创新想法
适应复杂性：能够处理需要多维度考量的复杂问题
自组织能力：团队成员可以动态调整角色和职责
基本用法示例
以下是创建和使用基本智能体团队的完整示例：
from langchain_openai import ChatOpenAI
from langchain.agents import Tool
from langchain.schema import SystemMessage, HumanMessage, AIMessage
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents.agent_toolkits import create_conversational_retrieval_agent
from langchain.schema.runnable import RunnablePassthrough
from typing import List, Dict, Any
# 初始化基础模型
llm = ChatOpenAI(temperature=0)
# 创建团队成员配置
team_members = {
"创意专家": {
"description": "专门负责创新思考和创意生成的专家。擅长提出新颖的想法和解决方案。",
"system_message": "你是一个创意思考专家，擅长产生创新的想法和解决方案。思考问题时，你应该考虑不寻常的角度，提出突破性的建议。"
},
"批判分析师": {
"description": "专门负责逻辑分析和批判性思考的专家。擅长评估想法的可行性和发现潜在问题。",
"system_message": "你是一个批判性思维专家，擅长逻辑分析和问题评估。你的任务是仔细评估提出的想法，找出潜在缺陷，并提供建设性的改进建议。"
},
"执行规划师": {
"description": "专门负责实施策略和行动计划的专家。擅长将想法转化为具体可行的步骤。",
"system_message": "你是一个执行规划专家，擅长创建实施方案和行动计划。你的任务是将概念性想法转化为具体、可行的步骤，并考虑资源、时间和可行性因素。"
}
}
# 创建团队协调者
class AgentTeamCoordinator:
def __init__(self, llm, team_members: Dict[str, Dict]):
self.llm = llm
self.team_members = team_members
self.conversation_history = []
def _create_team_member_message(self, role: str, content: str) -> dict:
"""创建团队成员消息格式"""
return {
"role": role,
"content": content
}
def get_team_member_input(self, member_name: str, task: str) -> str:
"""获取特定团队成员对任务的输入"""
member_config = self.team_members[member_name]
# 构建提示，包括之前的对话历史
messages = [
SystemMessage(content=member_config["system_message"]),
HumanMessage(content=f"任务: {task}\n\n请以{member_name}的角色，根据你的专业知识和视角，对这个任务提供你的见解和建议。")
]
# 添加对话历史供参考
if self.conversation_history:
history_text = "\n\n团队讨论历史:\n"
for msg in self.conversation_history:
history_text += f"\n{msg['role']}: {msg['content']}"
messages.append(HumanMessage(content=f"{history_text}\n\n基于以上历史和你的专业知识，请提供你的观点。"))
# 获取响应
response = self.llm.invoke(messages)
# 记录到对话历史
self.conversation_history.append(self._create_team_member_message(member_name, response.content))
return response.content
def synthesize_team_inputs(self, task: str, member_inputs: Dict[str, str]) -> str:
"""综合团队成员的输入，得出最终结论"""
synthesis_prompt = f"""作为团队协调者，你的任务是综合以下团队成员对任务的见解，提出一个全面的最终解决方案。
任务：{task}
团队成员输入：
"""
for member_name, input_text in member_inputs.items():
synthesis_prompt += f"\n\n{member_name} ({self.team_members[member_name]['description']}):\n{input_text}"
synthesis_prompt += "\n\n综合以上团队成员的见解，提出一个平衡各种观点、全面而实用的最终解决方案。解释您如何整合各成员的建议。"
response = self.llm.invoke([HumanMessage(content=synthesis_prompt)])
return response.content
def solve_task(self, task: str) -> str:
"""使用团队智能体解决任务"""
print(f"开始解决任务: {task}")
self.conversation_history = [] # 清空之前的对话历史
# 获取每个团队成员的输入
member_inputs = {}
for member_name in self.team_members.keys():
print(f"\n请求 {member_name} 的输入...")
member_inputs[member_name] = self.get_team_member_input(member_name, task)
print(f"{member_name} 回应: {member_inputs[member_name][:100]}...")
# 综合团队输入
print("\n综合团队输入...")
final_solution = self.synthesize_team_inputs(task, member_inputs)
return final_solution
# 创建团队并解决任务
team_coordinator = AgentTeamCoordinator(llm, team_members)
result = team_coordinator.solve_task("设计一个创新的智能家居系统，既要满足用户便利性需求，又要考虑隐私保护和能源效率。")
print("\n最终解决方案:\n")
print(result)
高级用法
动态团队讨论
以下示例展示了如何实现更复杂的团队讨论流程，允许团队成员进行多轮交流：
class DynamicAgentTeam:
def __init__(self, llm, team_members: Dict[str, Dict]):
self.llm = llm
self.team_members = team_members
self.discussion_history = []
def _format_discussion_history(self) -> str:
"""将讨论历史格式化为可读文本"""
if not self.discussion_history:
return ""
history = "\n\n讨论历史:\n"
for entry in self.discussion_history:
history += f"\n{entry['speaker']}: {entry['message']}"
return history
def team_member_speak(self, member_name: str, task: str, specific_question: str = "") -> str:
"""让特定团队成员发言"""
member_config = self.team_members[member_name]
# 构建提示
discussion_history = self._format_discussion_history()
if specific_question:
prompt = f"""任务: {task}
作为{member_name}，{member_config['system_message']}
{discussion_history}
现在请回应这个具体问题: {specific_question}
请基于你的专业知识提供见解，并考虑之前的讨论内容。
"""
else:
prompt = f"""任务: {task}
作为{member_name}，{member_config['system_message']}
{discussion_history}
考虑到讨论的当前状态，请提供你的专业见解。如果这是讨论的开始，请先分享你对任务的初步想法。
"""
response = self.llm.invoke([HumanMessage(content=prompt)])
# 记录到讨论历史
self.discussion_history.append({
"speaker": member_name,
"message": response.content
})
return response.content
def moderator_guide_discussion(self) -> dict:
"""由主持人引导讨论，确定下一步"""
discussion_history = self._format_discussion_history()
prompt = f"""作为团队讨论的主持人，你的任务是评估当前讨论状态，并决定下一步行动。
{discussion_history}
基于当前讨论，请判断:
1. 讨论是否应该继续？如果是，哪位团队成员应该发言，以及他们应该回应什么具体问题？
2. 如果讨论已充分，请说明原因并提出应该如何整合团队见解得出最终结论。
以JSON格式返回你的决定:
{{
"continue_discussion": true或false,
"next_speaker": "团队成员名称或'none'",
"question": "下一个讨论问题或''(如果讨论结束)",
"reasoning": "你的推理过程"
}}
"""
response = self.llm.invoke([HumanMessage(content=prompt)])
# 解析响应获取决策
# 实际应用中应添加错误处理
import json
try:
decision = json.loads(response.content)
return decision
except:
# 解析失败时的后备方案
return {
"continue_discussion": False,
"next_speaker": "none",
"question": "",
"reasoning": "无法解析决策，建议结束讨论并综合现有输入。"
}
def synthesize_final_solution(self, task: str) -> str:
"""综合讨论结果，形成最终解决方案"""
discussion_history = self._format_discussion_history()
prompt = f"""作为团队协调者，你的任务是综合团队讨论，提出一个全面的最终解决方案。
任务：{task}
{discussion_history}
请提供一个综合各团队成员观点的最终解决方案。确保解决方案是全面、平衡且可行的。
解释你如何整合各种观点，以及最终解决方案如何解决原始任务。
"""
response = self.llm.invoke([HumanMessage(content=prompt)])
return response.content
def facilitate_team_discussion(self, task: str, max_turns: int = 6) -> str:
"""组织团队讨论，解决任务"""
print(f"开始团队讨论: {task}")
self.discussion_history = [] # 清空之前的讨论
# 初始发言人
initial_speaker = list(self.team_members.keys())[0]
print(f"\n{initial_speaker} 开始讨论...")
self.team_member_speak(initial_speaker, task)
# 讨论轮次
turn = 1
while turn < max_turns:
# 主持人引导讨论
decision = self.moderator_guide_discussion()
print(f"\n讨论轮次 {turn} - 决策: {decision['reasoning'][:100]}...")
# 检查是否继续讨论
if not decision["continue_discussion"]:
print("讨论已充分，准备总结...")
break
# 让下一个发言人发言
next_speaker = decision["next_speaker"]
specific_question = decision["question"]
print(f"\n{next_speaker} 回应: {specific_question[:50]}...")
self.team_member_speak(next_speaker, task, specific_question)
turn += 1
# 综合讨论结果
print("\n综合讨论结果...")
final_solution = self.synthesize_final_solution(task)
return final_solution
# 使用动态团队讨论解决问题
dynamic_team = DynamicAgentTeam(llm, team_members)
complex_task = "设计一个可持续发展的城市交通系统，平衡效率、环境影响和社会公平性。"
result = dynamic_team.facilitate_team_discussion(complex_task)
print("\n最终解决方案:\n")
print(result)
以下是一个针对特定专业领域（如软件开发）的团队示例：
```python
from langchain.tools import tool
# 创建软件开发团队
software_team_members = {
"产品经理": {
"description": "负责定义产品需求和功能优先级",
"system_message": "你是一名经验丰富的产品经理，负责理解用户需求并将其转化为清晰的产品规范。你关注用户体验和业务目标。"
},
"架构师": {
"description": "负责技术架构设计和技术选型",
"system_message": "你是一名软件架构师，专注于设计可扩展、可维护的系统架构。你需要平衡技术债务、性能要求和实施复杂性。"
},
"开发工程师": {
"description": "负责实现具体功能和编写代码",
"system_message": "你是一名全栈开发工程师，擅长将需求转化为实际代码。你关注代码质量、可测试性和最佳实践。"
},
"QA测试员": {
"description": "负责质量保证和测试策略",
"system_message": "你是一名QA专家，负责确保软件质量。你擅长识别潜在问题、设计测试用例和验证功能完整性。"
}
}
# 为开发团队添加专业工具
@tool
def search_technical_docs(query: str) -> str:
"""搜索技术文档获取信息"""
# 实际应用中连接到技术文档数据库
return f"关于'{query}'的技术文档搜索结果：[相关技术信息]"
@tool
def analyze_code(code_snippet: str) -> str:
"""分析代码片段的质量和潜在问题"""
# 实际应用中连接到代码分析服务
return f"代码分析结果：[代码质量评估和建议]"
@tool
def estimate_effort(feature_description: str) -> str:
"""估算开发特性所需的工作量"""
# 实际应用中可能使用历史数据和ML模型
return f"工作量估算：基于'{feature_description}'的特性，预计需要[X]人天"
# 创建软件开发团队协调器的扩展类
class SoftwareDevelopmentTeam(DynamicAgentTeam):
def __init__(self, llm, team_members):
super().__init__(llm, team_members)
self.project_artifacts = {
"requirements": "",
"architecture": "",
"implementation_plan": "",
"test_strategy": ""
}
def create_artifact(self, artifact_type: str, content: str):
"""创建或更新项目工件"""
if artifact_type in self.project_artifacts:
self.project_artifacts[artifact_type] = content
print(f"\n✅ 已更新{artifact_type}文档")
return f"{artifact_type}已成功更新"
return f"未知工件类型：{artifact_type}"
def generate_project_documentation(self) -> str:
"""生成完整的项目文档"""
prompt = f"""基于以下项目工件，创建一个综合的项目文档：
需求文档：
{self.project_artifacts['requirements']}
架构设计：
{self.project_artifacts['architecture']}
实施计划：
{self.project_artifacts['implementation_plan']}
测试策略：
{self.project_artifacts['test_strategy']}
请创建一个结构清晰、内容全面的项目文档，包括概述、详细需求、技术架构、开发计划和质量保证措施。
"""
response = self.llm.invoke([HumanMessage(content=prompt)])
return response.content
def develop_software_project(self, project_description: str) -> str:
"""开发完整的软件项目"""
# 第一阶段：需求分析
print("\n📌 阶段1：需求分析")
requirement_task = f"针对以下项目进行需求分析：{project_description}\n请详细列出功能需求、非功能需求和用户场景。"
self.discussion_history = []
self.team_member_speak("产品经理", requirement_task)
self.create_artifact("requirements", self.discussion_history[-1]["message"])
# 第二阶段：架构设计
print("\n📌 阶段2：架构设计")
architecture_task = f"根据以下需求设计系统架构：\n{self.project_artifacts['requirements']}"
self.discussion_history = []
self.team_member_speak("架构师", architecture_task)
self.create_artifact("architecture", self.discussion_history[-1]["message"])
# 第三阶段：实施计划
print("\n📌 阶段3：实施计划")
implementation_task = f"根据以下需求和架构设计，制定实施计划：\n需求：{self.project_artifacts['requirements'][:200]}...\n架构：{self.project_artifacts['architecture'][:200]}..."
self.discussion_history = []
result = self.facilitate_team_discussion(implementation_task, max_turns=4)
self.create_artifact("implementation_plan", result)
# 第四阶段：测试策略
print("\n📌 阶段4：测试策略")
test_task = f"为以下项目设计测试策略：\n{project_description}\n考虑单元测试、集成测试和用户验收测试。"
self.discussion_history = []
self.team_member_speak("QA测试员", test_task)
self.create_artifact("test_strategy", self.discussion_history[-1]["message"])
# 最终项目文档
print("\n📌 生成最终项目文档")
final_documentation = self.generate_project_documentation()
return final_documentation
# 使用软件开发团队
software_team = SoftwareDevelopmentTeam(llm, software_team_members)
project_description = "开发一个在线团队协作工具，支持任务管理、文档共享和实时通信。系统需要高度可扩展，支持web和移动端访问。"
final_documentation = software_team.develop_software_project(project_description)
实际应用场景
1. 创意和内容策划
智能体团队在创意和内容策划领域可以提供多角度的思考和综合创意能力。
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
# 创建具有不同创意风格的团队
creative_team = {
"头脑风暴专家": {
"description": "专注于快速生成大量创意想法，不过滤",
"system_message": "你是一个头脑风暴专家，专注于快速生成大量创意。不要自我审查，即使想法看起来不寻常或不可行。你的目标是数量和多样性。"
},
"叙事专家": {
"description": "专注于故事结构和情感连接",
"system_message": "你是一个叙事和故事架构专家。专注于如何通过引人入胜的故事结构、人物塑造和情感连接来吸引目标受众。"
},
"趋势分析师": {
"description": "关注当前文化潮流和受众偏好",
"system_message": "你是一个文化趋势分析师，专注于识别和应用当前的文化趋势、热门话题和受众偏好。确保创意与当前环境相关并引起共鸣。"
},
"品牌策略师": {
"description": "确保创意符合品牌形象和营销目标",
"system_message": "你是一个品牌策略专家，确保创意符合品牌身份、价值观和营销目标。你关注创意如何加强品牌形象和与目标受众的长期关系。"
}
}
# 创建内容策划团队协调器
class ContentIdeationTeam(DynamicAgentTeam):
def __init__(self, llm, team_members):
super().__init__(llm, team_members)
def generate_content_concept(self, brief: str, target_audience: str, brand_guidelines: str) -> str:
"""生成内容创意概念"""
# 构建任务描述
task = f"""创建内容策划方案
项目简介：{brief}
目标受众：{target_audience}
品牌指南：{brand_guidelines}
请团队合作开发一个全面的内容创意概念，包括核心信息、叙事方法、创意元素和执行建议。"""
# 进行团队讨论
result = self.facilitate_team_discussion(task, max_turns=8)
return result
def evaluate_concept(self, concept: str, criteria: str) -> str:
"""评估创意概念"""
evaluation_task = f"""评估以下内容创意概念：
{concept}
评估标准：
{criteria}
请团队协作评估这个概念的优势、劣势和改进机会。提供具体、实用的反馈。"""
self.discussion_history = [] # 清空之前的讨论
result = self.facilitate_team_discussion(evaluation_task, max_turns=5)
return result
2. 决策支持系统
智能体团队可以作为高级决策支持系统，综合不同视角和专业知识来分析复杂问题。
from datetime import datetime
# 创建决策团队
decision_team = {
"数据分析师": {
"description": "专注于数据分析和量化评估",
"system_message": "你是一个数据分析专家，擅长解读数据，发现趋势和基于证据进行推理。你始终寻求量化证据和具体数字来支持决策。"
},
"风险评估师": {
"description": "专注于识别潜在风险和缓解策略",
"system_message": "你是一个风险评估专家，专注于识别方案中的潜在风险、挑战和意外后果。你应该考虑短期和长期风险，并提出缓解策略。"
},
"战略规划师": {
"description": "专注于长期战略影响和机会",
"system_message": "你是一个战略规划专家，专注于长期视角和战略影响。你考虑决策如何影响未来的增长、市场定位和竞争优势。"
},
"利益相关者代表": {
"description": "代表不同利益相关者的观点和需求",
"system_message": "你代表各种利益相关者的观点和需求，包括客户、员工、投资者、社区和监管机构。确保决策考虑到所有受影响方的利益和关切。"
}
}
# 创建决策支持团队协调器
class DecisionSupportTeam(DynamicAgentTeam):
def __init__(self, llm, team_members):
super().__init__(llm, team_members)
self.decision_matrix = {}
def analyze_options(self, scenario: str, options: list, criteria: list) -> str:
"""分析决策选项"""
# 构建分析任务
options_text = "\n".join([f"选项{i+1}: {option}" for i, option in enumerate(options)])
criteria_text = "\n".join([f"标准{i+1}: {criterion}" for i, criterion in enumerate(criteria)])
task = f"""决策场景分析：
场景：{scenario}
待评估选项：
{options_text}
评估标准：
{criteria_text}
请团队协作分析这些选项，考虑每个选项相对于评估标准的优缺点。提供全面、平衡的分析，并最终推荐最佳选择。
"""
result = self.facilitate_team_discussion(task, max_turns=10)
return result
def scenario_planning(self, context: str, scenarios: list, time_horizon: str) -> str:
"""进行情景规划分析"""
scenarios_text = "\n".join([f"情景{i+1}: {scenario}" for i, scenario in enumerate(scenarios)])
task = f"""情景规划分析：
当前背景：{context}
可能的未来情景：
{scenarios_text}
时间范围：{time_horizon}
请团队分析这些不同的未来情景，评估每种情景的可能性、影响和战略响应。对于每个情景，提供：
1. 关键驱动因素和早期警示信号
2. 潜在影响和风险
3. 建议的准备策略和应对措施
4. 情景间的共性机会和威胁
最后，提供一个综合战略，帮助在所有情景下保持适应性和韧性。
"""
result = self.facilitate_team_discussion(task, max_turns=12)
return result
最佳实践
明确角色定义：为每个团队成员定义明确的角色、专长和职责范围。
# 在团队成员定义中包含明确的角色边界 team_member_config = { "专家A": { "description": "...", "system_message": "你是专家A，专注于X领域。你的主要职责是Y。请不要涉及Z领域，那是其他团队成员的专长。", "strengths": ["优势1", "优势2"], "limitations": ["局限1", "局限2"] } }有效信息共享：确保团队成员之间能够有效共享和获取所需信息。
构建完整上下文：为每个成员提供足够的任务上下文和其他成员的相关输入。
平衡发言机会：确保所有团队成员都有机会贡献其专业知识，避免单个成员主导讨论。
有效冲突解决：建立冲突解决机制，处理团队成员之间的不同意见。
def resolve_conflict(topic: str, opinions: Dict[str, str]) -> str: """解决团队成员之间的冲突""" conflict_prompt = f"""作为中立的协调者，你需要帮助解决团队成员关于'{topic}'的不同观点： {format_opinions(opinions)} 请分析各种观点的优缺点，找出共同点，并提出一个平衡的解决方案。 说明你的推理过程和如何整合不同的观点。""" response = llm.invoke([HumanMessage(content=conflict_prompt)]) return response.content迭代改进：基于团队讨论和反馈不断改进解决方案。
适当的团队规模：根据任务复杂性选择合适的团队规模，通常保持在3-7名成员之间。
清晰的综合机制：建立明确的方法来综合各成员的贡献形成最终输出。
与其他多智能体模式的对比
常见问题与解决方案
角色重叠
解决方案：明确定义每个智能体的专长和边界
在提示中强调各自的独特视角
讨论发散
解决方案：实现有效的讨论引导机制
定期评估讨论进展并重新聚焦
信息过载
解决方案：实现有效的信息摘要和过滤机制
只共享与各成员角色相关的信息
决策延迟
解决方案：设置讨论轮次限制
实现决策触发机制，如达到共识阈值时
成员贡献不平衡
解决方案：实现参与监控机制
主动邀请贡献较少的成员发言
总结
智能体团队是一种强大的多智能体协作模式，特别适合需要多视角综合考量的复杂问题。通过组织具有不同专长和视角的智能体成员进行平等协作，团队可以产生比单个智能体更全面、更平衡的解决方案。
与其他多智能体模式相比，智能体团队注重成员之间的多向交流和共识形成，特别适用于创意生成、政策制定、复杂决策等需要综合多种视角的场景。然而，这种模式也需要更复杂的协调机制来管理团队交流和整合成果。
通过遵循本文档中的最佳实践和实施建议，开发者可以构建高效、协作的智能体团队系统，充分利用多智能体协作的优势解决复杂问题。
- 感谢你赐予我前进的力量
-
微信 
支付宝
赞赏者名单
因为你们的支持让我意识到写文章的价值🙏
本文是原创文章，采用 CC BY-NC-ND 4.0 协议，完整转载请注明来自 数据·开英
阅读建议
评论
匿名评论
隐私政策
你无需删除空行，直接评论以获取最佳展示效果
