在本案例中我们将探索如何创建一个多代理系统该系统使用顶级代理来协调一组作为工具的代理。具体来说我们将创建一个能够研究、撰写和审查关于给定主题报告的系统。1. 案例目标本案例的主要目标是演示如何使用LlamaIndex构建一个多代理系统其中子代理作为工具被顶级代理调用。具体来说我们将创建三个具有不同专业能力的子代理研究代理、写作代理和审查代理将这些子代理封装为工具使它们可以被顶级代理调用创建一个顶级协调代理负责协调子代理的执行顺序和交互实现状态共享使代理能够访问和修改共享数据展示如何通过流式输出监控多代理系统的执行过程2. 技术栈与核心依赖LlamaIndex- 用于构建多代理系统和工作流OpenAI API- 提供语言模型Tavily API- 提供网络搜索功能FunctionAgent- 用于构建可执行工具调用的代理Context- 用于在代理之间共享状态核心依赖包%pip install llama-index %pip install tavily-python3. 环境配置首先需要设置OpenAI API密钥from llama_index.llms.openai import OpenAI sub_agent_llm OpenAI(modelgpt-4.1-mini, api_keysk-...) orchestrator_llm OpenAI(modelo3-mini, api_keysk-...)4. 案例实现4.1 系统设计我们的系统将包含三个子代理ResearchAgent- 用于在网络上搜索给定主题的信息WriteAgent- 使用ResearchAgent找到的信息撰写报告ReviewAgent- 审查报告并提供反馈然后我们将使用一个顶级代理来协调其他代理撰写我们的报告。4.2 工具定义定义一个网络搜索工具from tavily import AsyncTavilyClient async def search_web(query: str) - str: Useful for using the web to answer questions. client AsyncTavilyClient(api_keytvly-...) return str(await client.search(query))4.3 子代理定义使用定义好的工具创建三个子代理from llama_index.core.agent.workflow import FunctionAgent, ReActAgent research_agent FunctionAgent( system_prompt( You are the ResearchAgent that can search the web for information on a given topic and record notes on the topic. You should output notes on the topic in a structured format. ), llmsub_agent_llm, tools[search_web], ) write_agent FunctionAgent( system_prompt( You are the WriteAgent that can write a report on a given topic. Your report should be in a markdown format. The content should be grounded in the research notes. Return your markdown report surrounded by ... tags. ), llmsub_agent_llm, tools[], ) review_agent FunctionAgent( system_prompt( You are the ReviewAgent that can review the write report and provide feedback. Your review should either approve the current report or request changes to be implemented. ), llmsub_agent_llm, tools[], )4.4 将子代理转换为工具将子代理封装为工具使它们可以被顶级代理调用import re from llama_index.core.workflow import Context async def call_research_agent(ctx: Context, prompt: str) - str: Useful for recording research notes based on a specific prompt. result await research_agent.run( user_msgfWrite some notes about the following: {prompt} ) async with ctx.store.edit_state() as ctx_state: ctx_state[state][research_notes].append(str(result)) return str(result) async def call_write_agent(ctx: Context) - str: Useful for writing a report based on the research notes or revising the report based on feedback. async with ctx.store.edit_state() as ctx_state: notes ctx_state[state].get(research_notes, None) if not notes: return No research notes to write from. user_msg fWrite a markdown report from the following notes. Be sure to output the report in the following format: ...:\n\n # Add the feedback to the user message if it exists feedback ctx_state[state].get(review, None) if feedback: user_msg f{feedback}\n\n # Add the research notes to the user message notes \n\n.join(notes) user_msg f{notes}\n\n # Run the write agent result await write_agent.run(user_msguser_msg) report re.search( r(.*), str(result), re.DOTALL ).group(1) ctx_state[state][report_content] str(report) return str(report) async def call_review_agent(ctx: Context) - str: Useful for reviewing the report and providing feedback. async with ctx.store.edit_state() as ctx_state: report ctx_state[state].get(report_content, None) if not report: return No report content to review. result await review_agent.run( user_msgfReview the following report: {report} ) ctx_state[state][review] result return result4.5 创建顶级协调代理使用封装好的工具创建顶级协调代理orchestrator FunctionAgent( system_prompt( You are an expert in the field of report writing. You are given a user request and a list of tools that can help with the request. You are to orchestrate the tools to research, write, and review a report on the given topic. Once the review is positive, you should notify the user that the report is ready to be accessed. ), llmorchestrator_llm, tools[ call_research_agent, call_write_agent, call_review_agent, ], initial_state{ research_notes: [], report_content: None, review: None, }, )4.6 运行代理运行代理并流式输出事件以便了解内部发生的情况from llama_index.core.agent.workflow import ( AgentInput, AgentOutput, ToolCall, ToolCallResult, AgentStream, ) from llama_index.core.workflow import Context # Create a context for the orchestrator to hold history/state ctx Context(orchestrator) async def run_orchestrator(ctx: Context, user_msg: str): handler orchestrator.run( user_msguser_msg, ctxctx, ) async for event in handler.stream_events(): if isinstance(event, AgentStream): if event.delta: print(event.delta, end, flushTrue) elif isinstance(event, AgentOutput): if event.tool_calls: print( ️ Planning to use tools:, [call.tool_name for call in event.tool_calls], ) elif isinstance(event, ToolCallResult): print(f Tool Result ({event.tool_name}):) print(f Arguments: {event.tool_kwargs}) print(f Output: {event.tool_output}) elif isinstance(event, ToolCall): print(f Calling Tool: {event.tool_name}) print(f With arguments: {event.tool_kwargs})4.7 执行示例await run_orchestrator( ctxctx, user_msg( Write me a report on the history of the internet. Briefly describe the history of the internet, including the development of the internet, the development of the web, and the development of the internet in the 21st century. ), )4.8 获取最终报告从系统状态中检索最终报告state await ctx.store.get(state) print(state[report_content])5. 案例效果5.1 多代理协作系统成功实现了三个代理的协作流程顶级协调代理首先调用研究代理搜索互联网历史相关信息研究代理返回研究笔记后协调代理调用写作代理写作代理基于研究笔记撰写报告协调代理调用审查代理审查报告审查代理提供反馈后协调代理再次调用写作代理修改报告最终生成高质量的报告5.2 代理作为工具通过将子代理封装为工具我们实现了代理的模块化和可重用性。每个子代理专注于特定任务而顶级代理负责协调这些任务的执行。这种设计使得系统更加灵活和可扩展。5.3 状态共享通过Context对象代理能够共享和修改状态。研究代理的研究笔记被写作代理用于撰写报告而写作代理撰写的报告被审查代理用于审查。这种状态共享机制使得代理能够协同工作共同完成任务。5.4 流式输出通过流式输出用户可以实时监控多代理系统的执行过程包括当前活动的代理、计划使用的工具、工具调用结果等。这种透明度使得系统行为更加可理解和可调试。5.5 迭代改进系统支持迭代改进过程。审查代理提供反馈后写作代理可以根据反馈修改报告直到报告满足要求。这种迭代机制确保了最终输出的高质量。6. 案例实现思路6.1 代理作为工具的设计模式本案例基于代理作为工具的设计模式将子代理封装为工具使它们可以被其他代理调用。这种模式的优势在于模块化每个代理专注于特定领域的任务提高执行质量可重用性代理可以在不同的工作流中重复使用层次化通过顶级代理协调子代理实现复杂的任务分解和执行6.2 代理协调机制顶级协调代理负责管理子代理的执行顺序和交互。它根据任务需求决定调用哪个子代理并处理子代理的返回结果。这种协调机制确保了工作流的有序执行和代理间的协作。6.3 状态共享Context对象提供了状态共享的机制使得代理能够访问和修改共享数据。通过Context.store.edit_state()方法代理可以安全地修改状态而其他代理可以访问这些状态。这种机制是实现代理协作的关键。6.4 工具封装通过将子代理封装为工具我们实现了代理的标准化接口。每个工具函数负责调用相应的子代理并处理返回结果。这种封装使得代理可以被其他代理无缝集成和使用。6.5 流式处理流式处理通过事件系统实现允许用户实时监控系统执行过程。系统会发出各种事件如AgentStream、ToolCall、ToolCallResult等用户可以监听这些事件并实时处理实现更好的用户体验和系统透明度。7. 扩展建议7.1 更复杂的代理网络设计更复杂的多层代理网络包含更多类型的代理和更复杂的交互模式实现代理间的直接通信机制使代理能够交换信息和协调行动添加动态代理创建和销毁功能使系统能够根据任务需求动态调整代理网络7.2 增强代理能力为代理添加更多专业工具如数据分析、图表生成、代码执行等实现代理学习能力使代理能够从过去的执行中学习和改进添加代理个性化设置使每个代理具有独特的风格和特点7.3 改进协调机制实现更智能的协调策略如基于任务优先级、代理负载和能力的动态调度添加并行执行支持使多个代理能够同时处理不同的子任务实现条件分支和循环使协调代理能够根据中间结果动态调整执行路径7.4 增强状态管理实现分层状态管理区分全局状态和局部状态添加状态版本控制和历史记录支持状态回滚和审计实现状态持久化和恢复使长时间运行的工作流能够从故障中恢复7.5 提高系统性能优化代理调用缓存减少重复的计算和网络请求实现智能调度策略根据代理负载和任务优先级动态分配资源添加性能监控和分析工具识别瓶颈并优化系统性能8. 总结本案例展示了如何使用LlamaIndex构建一个多代理系统其中子代理作为工具被顶级代理调用。通过这个案例我们了解了如何将子代理封装为工具实现代理的模块化和可重用性如何创建顶级协调代理负责管理子代理的执行顺序和交互如何使用Context对象在代理之间共享状态如何通过流式输出监控多代理系统的执行过程如何实现迭代改进机制确保最终输出的高质量代理作为工具的设计模式为构建复杂的AI应用提供了强大而灵活的框架。通过将复杂任务分解为多个子任务并让专门的代理负责每个子任务我们可以构建出更加智能、更加专业的系统。这种架构特别适合处理需要多种技能和知识的复杂任务。顶级协调代理作为系统的大脑负责管理和调度子代理的执行确保工作流的有序进行。这种层次化的代理架构使得系统更加灵活和可扩展能够适应各种复杂任务的需求。通过本案例的方法开发者可以快速构建各种多代理系统实现复杂任务的自动化处理。这些系统不仅能够提高工作效率还能够提供更加专业和高质量的结果为各行各业的数字化转型提供支持。
在本案例中我们将探索如何创建一个多代理系统该系统使用顶级代理来协调一组作为工具的代理。具体来说我们将创建一个能够研究、撰写和审查关于给定主题报告的系统。1. 案例目标本案例的主要目标是演示如何使用LlamaIndex构建一个多代理系统其中子代理作为工具被顶级代理调用。具体来说我们将创建三个具有不同专业能力的子代理研究代理、写作代理和审查代理将这些子代理封装为工具使它们可以被顶级代理调用创建一个顶级协调代理负责协调子代理的执行顺序和交互实现状态共享使代理能够访问和修改共享数据展示如何通过流式输出监控多代理系统的执行过程2. 技术栈与核心依赖LlamaIndex- 用于构建多代理系统和工作流OpenAI API- 提供语言模型Tavily API- 提供网络搜索功能FunctionAgent- 用于构建可执行工具调用的代理Context- 用于在代理之间共享状态核心依赖包%pip install llama-index %pip install tavily-python3. 环境配置首先需要设置OpenAI API密钥from llama_index.llms.openai import OpenAI sub_agent_llm OpenAI(modelgpt-4.1-mini, api_keysk-...) orchestrator_llm OpenAI(modelo3-mini, api_keysk-...)4. 案例实现4.1 系统设计我们的系统将包含三个子代理ResearchAgent- 用于在网络上搜索给定主题的信息WriteAgent- 使用ResearchAgent找到的信息撰写报告ReviewAgent- 审查报告并提供反馈然后我们将使用一个顶级代理来协调其他代理撰写我们的报告。4.2 工具定义定义一个网络搜索工具from tavily import AsyncTavilyClient async def search_web(query: str) - str: Useful for using the web to answer questions. client AsyncTavilyClient(api_keytvly-...) return str(await client.search(query))4.3 子代理定义使用定义好的工具创建三个子代理from llama_index.core.agent.workflow import FunctionAgent, ReActAgent research_agent FunctionAgent( system_prompt( You are the ResearchAgent that can search the web for information on a given topic and record notes on the topic. You should output notes on the topic in a structured format. ), llmsub_agent_llm, tools[search_web], ) write_agent FunctionAgent( system_prompt( You are the WriteAgent that can write a report on a given topic. Your report should be in a markdown format. The content should be grounded in the research notes. Return your markdown report surrounded by ... tags. ), llmsub_agent_llm, tools[], ) review_agent FunctionAgent( system_prompt( You are the ReviewAgent that can review the write report and provide feedback. Your review should either approve the current report or request changes to be implemented. ), llmsub_agent_llm, tools[], )4.4 将子代理转换为工具将子代理封装为工具使它们可以被顶级代理调用import re from llama_index.core.workflow import Context async def call_research_agent(ctx: Context, prompt: str) - str: Useful for recording research notes based on a specific prompt. result await research_agent.run( user_msgfWrite some notes about the following: {prompt} ) async with ctx.store.edit_state() as ctx_state: ctx_state[state][research_notes].append(str(result)) return str(result) async def call_write_agent(ctx: Context) - str: Useful for writing a report based on the research notes or revising the report based on feedback. async with ctx.store.edit_state() as ctx_state: notes ctx_state[state].get(research_notes, None) if not notes: return No research notes to write from. user_msg fWrite a markdown report from the following notes. Be sure to output the report in the following format: ...:\n\n # Add the feedback to the user message if it exists feedback ctx_state[state].get(review, None) if feedback: user_msg f{feedback}\n\n # Add the research notes to the user message notes \n\n.join(notes) user_msg f{notes}\n\n # Run the write agent result await write_agent.run(user_msguser_msg) report re.search( r(.*), str(result), re.DOTALL ).group(1) ctx_state[state][report_content] str(report) return str(report) async def call_review_agent(ctx: Context) - str: Useful for reviewing the report and providing feedback. async with ctx.store.edit_state() as ctx_state: report ctx_state[state].get(report_content, None) if not report: return No report content to review. result await review_agent.run( user_msgfReview the following report: {report} ) ctx_state[state][review] result return result4.5 创建顶级协调代理使用封装好的工具创建顶级协调代理orchestrator FunctionAgent( system_prompt( You are an expert in the field of report writing. You are given a user request and a list of tools that can help with the request. You are to orchestrate the tools to research, write, and review a report on the given topic. Once the review is positive, you should notify the user that the report is ready to be accessed. ), llmorchestrator_llm, tools[ call_research_agent, call_write_agent, call_review_agent, ], initial_state{ research_notes: [], report_content: None, review: None, }, )4.6 运行代理运行代理并流式输出事件以便了解内部发生的情况from llama_index.core.agent.workflow import ( AgentInput, AgentOutput, ToolCall, ToolCallResult, AgentStream, ) from llama_index.core.workflow import Context # Create a context for the orchestrator to hold history/state ctx Context(orchestrator) async def run_orchestrator(ctx: Context, user_msg: str): handler orchestrator.run( user_msguser_msg, ctxctx, ) async for event in handler.stream_events(): if isinstance(event, AgentStream): if event.delta: print(event.delta, end, flushTrue) elif isinstance(event, AgentOutput): if event.tool_calls: print( ️ Planning to use tools:, [call.tool_name for call in event.tool_calls], ) elif isinstance(event, ToolCallResult): print(f Tool Result ({event.tool_name}):) print(f Arguments: {event.tool_kwargs}) print(f Output: {event.tool_output}) elif isinstance(event, ToolCall): print(f Calling Tool: {event.tool_name}) print(f With arguments: {event.tool_kwargs})4.7 执行示例await run_orchestrator( ctxctx, user_msg( Write me a report on the history of the internet. Briefly describe the history of the internet, including the development of the internet, the development of the web, and the development of the internet in the 21st century. ), )4.8 获取最终报告从系统状态中检索最终报告state await ctx.store.get(state) print(state[report_content])5. 案例效果5.1 多代理协作系统成功实现了三个代理的协作流程顶级协调代理首先调用研究代理搜索互联网历史相关信息研究代理返回研究笔记后协调代理调用写作代理写作代理基于研究笔记撰写报告协调代理调用审查代理审查报告审查代理提供反馈后协调代理再次调用写作代理修改报告最终生成高质量的报告5.2 代理作为工具通过将子代理封装为工具我们实现了代理的模块化和可重用性。每个子代理专注于特定任务而顶级代理负责协调这些任务的执行。这种设计使得系统更加灵活和可扩展。5.3 状态共享通过Context对象代理能够共享和修改状态。研究代理的研究笔记被写作代理用于撰写报告而写作代理撰写的报告被审查代理用于审查。这种状态共享机制使得代理能够协同工作共同完成任务。5.4 流式输出通过流式输出用户可以实时监控多代理系统的执行过程包括当前活动的代理、计划使用的工具、工具调用结果等。这种透明度使得系统行为更加可理解和可调试。5.5 迭代改进系统支持迭代改进过程。审查代理提供反馈后写作代理可以根据反馈修改报告直到报告满足要求。这种迭代机制确保了最终输出的高质量。6. 案例实现思路6.1 代理作为工具的设计模式本案例基于代理作为工具的设计模式将子代理封装为工具使它们可以被其他代理调用。这种模式的优势在于模块化每个代理专注于特定领域的任务提高执行质量可重用性代理可以在不同的工作流中重复使用层次化通过顶级代理协调子代理实现复杂的任务分解和执行6.2 代理协调机制顶级协调代理负责管理子代理的执行顺序和交互。它根据任务需求决定调用哪个子代理并处理子代理的返回结果。这种协调机制确保了工作流的有序执行和代理间的协作。6.3 状态共享Context对象提供了状态共享的机制使得代理能够访问和修改共享数据。通过Context.store.edit_state()方法代理可以安全地修改状态而其他代理可以访问这些状态。这种机制是实现代理协作的关键。6.4 工具封装通过将子代理封装为工具我们实现了代理的标准化接口。每个工具函数负责调用相应的子代理并处理返回结果。这种封装使得代理可以被其他代理无缝集成和使用。6.5 流式处理流式处理通过事件系统实现允许用户实时监控系统执行过程。系统会发出各种事件如AgentStream、ToolCall、ToolCallResult等用户可以监听这些事件并实时处理实现更好的用户体验和系统透明度。7. 扩展建议7.1 更复杂的代理网络设计更复杂的多层代理网络包含更多类型的代理和更复杂的交互模式实现代理间的直接通信机制使代理能够交换信息和协调行动添加动态代理创建和销毁功能使系统能够根据任务需求动态调整代理网络7.2 增强代理能力为代理添加更多专业工具如数据分析、图表生成、代码执行等实现代理学习能力使代理能够从过去的执行中学习和改进添加代理个性化设置使每个代理具有独特的风格和特点7.3 改进协调机制实现更智能的协调策略如基于任务优先级、代理负载和能力的动态调度添加并行执行支持使多个代理能够同时处理不同的子任务实现条件分支和循环使协调代理能够根据中间结果动态调整执行路径7.4 增强状态管理实现分层状态管理区分全局状态和局部状态添加状态版本控制和历史记录支持状态回滚和审计实现状态持久化和恢复使长时间运行的工作流能够从故障中恢复7.5 提高系统性能优化代理调用缓存减少重复的计算和网络请求实现智能调度策略根据代理负载和任务优先级动态分配资源添加性能监控和分析工具识别瓶颈并优化系统性能8. 总结本案例展示了如何使用LlamaIndex构建一个多代理系统其中子代理作为工具被顶级代理调用。通过这个案例我们了解了如何将子代理封装为工具实现代理的模块化和可重用性如何创建顶级协调代理负责管理子代理的执行顺序和交互如何使用Context对象在代理之间共享状态如何通过流式输出监控多代理系统的执行过程如何实现迭代改进机制确保最终输出的高质量代理作为工具的设计模式为构建复杂的AI应用提供了强大而灵活的框架。通过将复杂任务分解为多个子任务并让专门的代理负责每个子任务我们可以构建出更加智能、更加专业的系统。这种架构特别适合处理需要多种技能和知识的复杂任务。顶级协调代理作为系统的大脑负责管理和调度子代理的执行确保工作流的有序进行。这种层次化的代理架构使得系统更加灵活和可扩展能够适应各种复杂任务的需求。通过本案例的方法开发者可以快速构建各种多代理系统实现复杂任务的自动化处理。这些系统不仅能够提高工作效率还能够提供更加专业和高质量的结果为各行各业的数字化转型提供支持。
