尽管单个自主智能体能够完成令人赞叹的任务,但其能力常受限于其单一的“思考路径”。对于单个智能体而言过于庞大或复杂的问题,您可以协调一个智能体团队,它们共同努力以达成一个共同目标。这种方法类似人类团队的工作方式,不同成员各司其职,并通过沟通解决问题。多智能体系统让您可以将一个复杂任务分解为更小、更易处理的子任务,并将每个子任务分配给一个专业智能体。例如,您可以不再使用单个“研究助理”智能体,而是组建一个团队,包含负责查找信息的研究员、负责起草内容的撰稿人和负责修改润色的编辑。kerb.agent模块提供了构建和管理这些协作系统的工具。核心组件是AgentTeam类,它协调多智能体之间的交互。创建智能体团队AgentTeam是一组独立的智能体实例。团队中的每个智能体都可以有自己独特的提示、工具集和配置,从而让您能够定义专业化的角色。我们首先定义几个专业智能体,并将它们组成一个团队。from kerb.agent.patterns import ReActAgent as Agent from kerb.agent.teams import AgentTeam # 为每个智能体的专业功能定义模拟的LLM函数 def researcher_llm(prompt: str) -> str: """研究员智能体的LLM。""" return "我找到了以下信息:Python是一种流行的编程语言。" def writer_llm(prompt: str) -> str: """撰稿人智能体的LLM。""" return "我根据研究结果撰写了一篇全面的文章。" def editor_llm(prompt: str) -> str: """编辑智能体的LLM。""" return "我已审阅并编辑了内容。现在它已润色完毕,可以使用了。" # 创建具有专业角色的独立智能体 researcher = Agent(name="Researcher", llm_func=researcher_llm) writer = Agent(name="Writer", llm_func=writer_llm) editor = Agent(name="Editor", llm_func=editor_llm) # 创建一个AgentTeam writing_team = AgentTeam(agents=[researcher, writer, editor]) print(f"已创建包含{len(writing_team.agents)}个智能体的团队。")组建好团队后,您可以运用不同的模式来协调它们的协作,例如顺序管道或并行工作流。管道的顺序执行一种常见的多智能体模式是顺序管道,智能体按照预设的顺序工作,如同装配线一般。一个智能体的输出成为下一个智能体的输入。这对于结构化的多步骤任务很有用,例如我们的研究和撰写例子。digraph G { rankdir=TB; node [shape=box, style="rounded", fontname="sans-serif", color="#4263eb", fontcolor="#4263eb"]; edge [color="#adb5bd"]; Goal -> Researcher [label="初始任务"]; Researcher -> Writer [label="研究成果"]; Writer -> Editor [label="文章草稿"]; Editor -> FinalResult [label="润色后的文章"]; FinalResult [shape=ellipse, color="#37b24d", fontcolor="#37b24d"]; }一个顺序管道,其中每个智能体的输出都会传递给下一个智能体。run_sequential方法按照智能体被添加到团队的顺序执行它们。整个过程的最终输出是序列中最后一个智能体的输出。goal_sequential = "撰写一篇关于Python的文章" print(f"目标:{goal_sequential}") print("正在顺序执行智能体(研究员 -> 撰稿人 -> 编辑)...") sequential_results = writing_team.run_sequential(goal_sequential) # 显示每一步的结果 for i, result in enumerate(sequential_results): agent_name = writing_team.agents[i-1].name print(f"\n[步骤 {i}] {agent_name}:") print(f" 输出:{result.output[:100]}...") if i < len(sequential_results): print(f" -> 已传递给下一个智能体")这种管道结构确保了信息流清晰顺畅,并且对于阶段明确的过程非常有效。并行执行与结果汇总另一种方法是并行执行,即多个智能体同时处理同一目标。这种模式对于收集不同观点或针对同一问题的解决方案很有用。每个智能体独立完成任务,它们的结果可以被审阅或汇总,以形成更全面的最终输出。digraph G { rankdir=TB; node [shape=box, style="rounded", fontname="sans-serif", color="#4263eb", fontcolor="#4263eb"]; edge [color="#adb5bd"]; subgraph cluster_workers { label="并行工作者"; bgcolor="#e9ecef"; border-style="dashed"; Researcher; Writer; Editor; } Goal -> {Researcher, Writer, Editor} [label="相同任务"]; {Researcher, Writer, Editor} -> Aggregator; Aggregator -> FinalResult [label="合并输出"]; FinalResult [shape=ellipse, color="#37b24d", fontcolor="#37b24d"]; }一个并行工作流,其中多个智能体处理同一目标。run_parallel方法将目标广播给团队中的所有智能体,并收集它们各自的结果。from kerb.agent.teams import aggregate_results goal_parallel = "研究使用Python进行AI的好处" print(f"目标:{goal_parallel}") parallel_results = writing_team.run_parallel(goal_parallel) print("\n各自结果:") for i, result in enumerate(parallel_results): agent_name = writing_team.agents[i-1].name print(f" - {agent_name}:{result.output[:60]}...") # 将各自的输出汇总为一个结果 aggregated_result = aggregate_results(parallel_results) print("\n汇总结果:") print(f" 合并输出:{aggregated_result.output[:100]}...")aggregate_results函数将所有智能体的输出合并为一个AgentResult对象。这使您能够展示所有观点,或将合并后的结果传递给另一个智能体进行综合。分层委派更为复杂的系统常采用分层结构,其中“经理”或“协调者”智能体分解一个大问题,并将子任务委派给专业“工作者”智能体。delegate_task函数方便了这种直接的智能体到智能体的任务分配。from kerb.agent.teams import delegate_task # 定义另一个专业智能体 def analyst_llm(prompt: str) -> str: return "分析完成:数据显示出积极趋势。" analyst = Agent(name="Analyst", llm_func=analyst_llm) print(f"\n{researcher.name} 向 {analyst.name} 委派任务") delegated_task = "分析研究成果中的市场趋势" delegation_result = delegate_task( task=delegated_task, from_agent=researcher, to_agent=analyst, context={'source': 'research_data.csv'} ) print(f"\n委派完成:") print(f" 任务:{delegated_task}") print(f" 来自 {analyst.name} 的结果:{delegation_result.output}")委派实现了动态且自适应的工作流,在此工作流中,智能体可以根据主任务不断变化的状态向其他智能体寻求帮助。带有对话历史的沟通为了智能体能有效协作,它们需要一种沟通和共享状态的方式。Conversation类充当共享消息日志,让智能体能够查看交互历史。这种共享背景有助于做出明智的决策。from kerb.agent.teams import Conversation # 团队共享的对话 conversation = Conversation() # 模拟智能体之间的对话 conversation.add_message(researcher.name, "我已完成关于Python在AI中的研究。") conversation.add_message(writer.name, "太好了!我将用它来撰写文章。") conversation.add_message(writer.name, "文章已起草完成,可以审阅了。") conversation.add_message(editor.name, "我现在就审阅并提供反馈。") print("\n对话历史:") for msg in conversation.get_history(): print(f" [{msg['timestamp']}] {msg['agent']}:{msg['content']}")这种对话历史可以被包含在智能体的上下文中,使其能够了解已经讨论和决定的内容。通过结合专业智能体、清晰的协调模式以及共享的通信渠道,您可以构建精巧的多智能体系统,它们能够解决单个智能体无法企及的问题。