尧图网页设计 
1. 案例目标本案例旨在展示如何使用LlamaIndex构建一个具有规划能力的多智能体系统该系统能够自动编写、优化和执行计划以生成高质量的报告。通过这个案例用户将了解如何创建多个专门的智能体研究、写作、评审并使用顶层规划器来协调这些智能体的工作流程实现从研究到最终报告的完整自动化流程。2. 技术栈与核心依赖核心技术LlamaIndex Workflow框架FunctionAgent智能体自定义工作流设计状态管理与持久化事件驱动架构流式输出处理依赖库llama-indextavily-python (网络搜索)pydantic (数据模型)xml.etree.ElementTree (XML解析)3. 环境配置# 安装必要的库 %pip install llama-index %pip install tavily-python # 配置LLM from llama_index.llms.openai import OpenAI sub_agent_llm OpenAI(modelgpt-4.1-mini, api_keysk-...) # 配置网络搜索工具 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))环境配置包括安装LlamaIndex和Tavily Python库配置OpenAI的GPT-4.1-mini模型作为子智能体的LLM以及设置Tavily作为网络搜索工具。4. 案例实现4.1 系统设计系统包含三个专门的智能体ResearchAgent负责搜索网络信息并记录研究笔记WriteAgent负责根据研究笔记撰写报告ReviewAgent负责评审报告并提供反馈顶层规划器使用LLM手动协调和规划这些智能体的工作流程以生成最终报告。4.2 创建子智能体from llama_index.core.agent.workflow import FunctionAgent research_agent FunctionAgent( nameResearchAgent, descriptionUseful for recording research notes based on a specific prompt., 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( nameWriteAgent, descriptionUseful for writing a report based on the research notes or revising the report based on feedback., 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( nameReviewAgent, descriptionUseful for reviewing a report and providing feedback., 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.3 智能体辅助函数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.4 定义规划器工作流创建自定义工作流来显式协调和规划其他智能体import re import xml.etree.ElementTree as ET from pydantic import BaseModel, Field from typing import Any, Optional from llama_index.core.llms import ChatMessage from llama_index.core.workflow import ( Context, Event, StartEvent, StopEvent, Workflow, step, ) PLANNER_PROMPT You are a planner chatbot. Given a user request and the current state, break the solution into ordered blocks. Each step must specify the agent to call and the message to send, e.g. search for … draft a report … ... {state} {available_agents} The general flow should be: - Record research notes - Write a report - Review the report - Write the report again if the review is not positive enough If the user request does not require any steps, you can skip the block and respond directly. 4.5 事件模型定义class InputEvent(StartEvent): user_msg: Optional[str] Field(defaultNone) chat_history: list[ChatMessage] state: Optional[dict[str, Any]] Field(defaultNone) class OutputEvent(StopEvent): response: str chat_history: list[ChatMessage] state: dict[str, Any] class StreamEvent(Event): delta: str class PlanEvent(Event): step_info: str # Modelling the plan class PlanStep(BaseModel): agent_name: str agent_input: str class Plan(BaseModel): steps: list[PlanStep] class ExecuteEvent(Event): plan: Plan chat_history: list[ChatMessage]4.6 规划器工作流实现class PlannerWorkflow(Workflow): llm: OpenAI OpenAI( modelo3-mini, api_keysk-..., ) agents: dict[str, FunctionAgent] { ResearchAgent: research_agent, WriteAgent: write_agent, ReviewAgent: review_agent, } step async def plan( self, ctx: Context, ev: InputEvent ) - ExecuteEvent | OutputEvent: # Set initial state if it exists if ev.state: await ctx.store.set(state, ev.state) chat_history ev.chat_history if ev.user_msg: user_msg ChatMessage( roleuser, contentev.user_msg, ) chat_history.append(user_msg) # Inject the system prompt with state and available agents state await ctx.store.get(state) available_agents_str \n.join( [ f{agent.description} for agent in self.agents.values() ] ) system_prompt ChatMessage( rolesystem, contentPLANNER_PROMPT.format( statestr(state), available_agentsavailable_agents_str, ), ) # Stream the response from the llm response await self.llm.astream_chat( messages[system_prompt] chat_history, ) full_response async for chunk in response: full_response chunk.delta or if chunk.delta: ctx.write_event_to_stream( StreamEvent(deltachunk.delta), ) # Parse the response into a plan and decide whether to execute or output xml_match re.search(r(.*), full_response, re.DOTALL) if not xml_match: chat_history.append( ChatMessage( roleassistant, contentfull_response, ) ) return OutputEvent( responsefull_response, chat_historychat_history, statestate, ) else: xml_str xml_match.group(1) root ET.fromstring(xml_str) plan Plan(steps[]) for step in root.findall(step): plan.steps.append( PlanStep( agent_namestep.attrib[agent], agent_inputstep.text.strip() if step.text else , ) ) return ExecuteEvent(planplan, chat_historychat_history)4.7 执行步骤实现step async def execute(self, ctx: Context, ev: ExecuteEvent) - InputEvent: chat_history ev.chat_history plan ev.plan for step in plan.steps: agent self.agents[step.agent_name] agent_input step.agent_input ctx.write_event_to_stream( PlanEvent( step_infof{step.agent_input} ), ) if step.agent_name ResearchAgent: await call_research_agent(ctx, agent_input) elif step.agent_name WriteAgent: # Note: we arent passing the input from the plan since # were using the state to drive the write agent await call_write_agent(ctx) elif step.agent_name ReviewAgent: await call_review_agent(ctx) state await ctx.store.get(state) chat_history.append( ChatMessage( roleuser, contentfIve completed the previous steps, heres the updated state:\n\n\n{state}\n\n\nDo you need to continue and plan more steps?, If not, write a final response., ) ) return InputEvent( chat_historychat_history, )4.8 运行工作流planner_workflow PlannerWorkflow(timeoutNone) handler planner_workflow.run( 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. ), chat_history[], state{ research_notes: [], report_content: Not written yet., review: Review required., }, ) current_agent None current_tool_calls async for event in handler.stream_events(): if isinstance(event, PlanEvent): print(Executing plan step: , event.step_info) elif isinstance(event, ExecuteEvent): print(Executing plan: , event.plan) result await handler4.9 获取最终报告print(result.response) # 获取最终报告内容 state await handler.ctx.store.get(state) print(state[report_content]) # 获取评审反馈 print(state[review])5. 案例效果通过自定义规划多智能体系统我们成功实现了以下功能自动规划多步骤任务流程将复杂报告生成任务分解为研究、写作和评审三个阶段ResearchAgent能够搜索网络信息并记录结构化的研究笔记WriteAgent能够根据研究笔记生成格式化的Markdown报告ReviewAgent能够评审报告并提供具体的改进建议规划器能够根据评审结果自动规划额外的修订步骤通过流式输出实时展示工作流执行过程状态管理确保各智能体之间的信息传递和持久化工作流程示例步骤1ResearchAgent - 记录关于互联网历史的研究笔记包括互联网发展、万维网出现和21世纪进展↓步骤2WriteAgent - 根据记录的研究笔记起草关于互联网历史的报告包括早期发展、万维网创建和影响以及21世纪里程碑↓步骤3ReviewAgent - 评审生成的报告并提供反馈确保报告符合请求标准↓步骤4WriteAgent - 根据评审建议修订报告包括澄清IETF角色、添加TCP/IP转换日期、解释浏览器战争等最终生成的报告包含了互联网发展的完整历史从ARPANET到现代互联网从万维网的诞生到21世纪的技术进步结构清晰内容详实并包含了权威参考文献。6. 案例实现思路1. 智能体分工设计将报告生成任务分解为三个专门的智能体研究、写作和评审每个智能体负责特定阶段的任务实现职责分离和专业化处理。2. 状态管理机制使用Context.store实现状态持久化确保各智能体之间的信息传递和共享。状态包括研究笔记、报告内容和评审反馈作为智能体之间协作的桥梁。3. 规划器设计创建自定义工作流PlannerWorkflow使用LLM根据用户请求和当前状态生成执行计划。计划以XML格式表示包含有序的步骤列表每个步骤指定要调用的智能体和输入消息。4. 事件驱动架构定义多种事件类型InputEvent、OutputEvent、StreamEvent、PlanEvent、ExecuteEvent实现工作流内部和与外部通信的事件驱动机制支持流式输出和实时状态更新。5. 迭代优化流程通过评审智能体的反馈机制实现报告的迭代优化。当评审不通过时规划器会自动生成新的修订步骤指导写作智能体根据反馈进行改进直到满足质量要求。7. 扩展建议功能扩展添加更多专业智能体如数据分析师、图表生成器实现并行任务执行提高处理效率添加多语言支持实现多语言报告生成集成更多数据源如学术数据库、内部文档实现报告模板系统支持不同格式和风格智能增强实现更智能的规划算法支持条件分支和循环添加智能体间协商机制解决冲突和优化协作实现自动质量评估减少人工评审需求添加学习能力从历史执行中优化规划策略实现用户偏好学习个性化报告风格用户体验添加可视化界面直观展示工作流执行状态实现交互式规划允许用户调整和确认计划添加实时预览功能展示报告生成过程实现版本控制支持报告历史查看和回滚添加协作功能支持多用户共同编辑集成与部署与企业系统集成如CRM、ERP系统实现API接口供其他系统调用添加权限管理和安全控制实现分布式部署提高系统可扩展性添加监控和日志系统便于运维管理8. 总结本案例展示了如何使用LlamaIndex构建一个具有规划能力的多智能体系统实现了从研究到最终报告的完整自动化流程。通过将复杂任务分解为多个专门智能体并使用顶层规划器协调它们的工作我们创建了一个灵活、可扩展的系统架构。案例中的关键技术点包括智能体分工设计、状态管理机制、规划器设计、事件驱动架构和迭代优化流程。这些技术的组合使系统能够自动处理复杂的多步骤任务并根据反馈进行自我优化。这种多智能体规划系统不仅适用于报告生成还可以扩展到其他复杂任务领域如项目管理、产品开发、科研流程等。通过添加更多专业智能体和优化规划算法可以进一步增强系统的能力和适用范围。 【Agent】11. 自定义规划多智能体系统案例分析
1. 案例目标本案例旨在展示如何使用LlamaIndex构建一个具有规划能力的多智能体系统该系统能够自动编写、优化和执行计划以生成高质量的报告。通过这个案例用户将了解如何创建多个专门的智能体研究、写作、评审并使用顶层规划器来协调这些智能体的工作流程实现从研究到最终报告的完整自动化流程。2. 技术栈与核心依赖核心技术LlamaIndex Workflow框架FunctionAgent智能体自定义工作流设计状态管理与持久化事件驱动架构流式输出处理依赖库llama-indextavily-python (网络搜索)pydantic (数据模型)xml.etree.ElementTree (XML解析)3. 环境配置# 安装必要的库 %pip install llama-index %pip install tavily-python # 配置LLM from llama_index.llms.openai import OpenAI sub_agent_llm OpenAI(modelgpt-4.1-mini, api_keysk-...) # 配置网络搜索工具 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))环境配置包括安装LlamaIndex和Tavily Python库配置OpenAI的GPT-4.1-mini模型作为子智能体的LLM以及设置Tavily作为网络搜索工具。4. 案例实现4.1 系统设计系统包含三个专门的智能体ResearchAgent负责搜索网络信息并记录研究笔记WriteAgent负责根据研究笔记撰写报告ReviewAgent负责评审报告并提供反馈顶层规划器使用LLM手动协调和规划这些智能体的工作流程以生成最终报告。4.2 创建子智能体from llama_index.core.agent.workflow import FunctionAgent research_agent FunctionAgent( nameResearchAgent, descriptionUseful for recording research notes based on a specific prompt., 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( nameWriteAgent, descriptionUseful for writing a report based on the research notes or revising the report based on feedback., 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( nameReviewAgent, descriptionUseful for reviewing a report and providing feedback., 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.3 智能体辅助函数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.4 定义规划器工作流创建自定义工作流来显式协调和规划其他智能体import re import xml.etree.ElementTree as ET from pydantic import BaseModel, Field from typing import Any, Optional from llama_index.core.llms import ChatMessage from llama_index.core.workflow import ( Context, Event, StartEvent, StopEvent, Workflow, step, ) PLANNER_PROMPT You are a planner chatbot. Given a user request and the current state, break the solution into ordered blocks. Each step must specify the agent to call and the message to send, e.g. search for … draft a report … ... {state} {available_agents} The general flow should be: - Record research notes - Write a report - Review the report - Write the report again if the review is not positive enough If the user request does not require any steps, you can skip the block and respond directly. 4.5 事件模型定义class InputEvent(StartEvent): user_msg: Optional[str] Field(defaultNone) chat_history: list[ChatMessage] state: Optional[dict[str, Any]] Field(defaultNone) class OutputEvent(StopEvent): response: str chat_history: list[ChatMessage] state: dict[str, Any] class StreamEvent(Event): delta: str class PlanEvent(Event): step_info: str # Modelling the plan class PlanStep(BaseModel): agent_name: str agent_input: str class Plan(BaseModel): steps: list[PlanStep] class ExecuteEvent(Event): plan: Plan chat_history: list[ChatMessage]4.6 规划器工作流实现class PlannerWorkflow(Workflow): llm: OpenAI OpenAI( modelo3-mini, api_keysk-..., ) agents: dict[str, FunctionAgent] { ResearchAgent: research_agent, WriteAgent: write_agent, ReviewAgent: review_agent, } step async def plan( self, ctx: Context, ev: InputEvent ) - ExecuteEvent | OutputEvent: # Set initial state if it exists if ev.state: await ctx.store.set(state, ev.state) chat_history ev.chat_history if ev.user_msg: user_msg ChatMessage( roleuser, contentev.user_msg, ) chat_history.append(user_msg) # Inject the system prompt with state and available agents state await ctx.store.get(state) available_agents_str \n.join( [ f{agent.description} for agent in self.agents.values() ] ) system_prompt ChatMessage( rolesystem, contentPLANNER_PROMPT.format( statestr(state), available_agentsavailable_agents_str, ), ) # Stream the response from the llm response await self.llm.astream_chat( messages[system_prompt] chat_history, ) full_response async for chunk in response: full_response chunk.delta or if chunk.delta: ctx.write_event_to_stream( StreamEvent(deltachunk.delta), ) # Parse the response into a plan and decide whether to execute or output xml_match re.search(r(.*), full_response, re.DOTALL) if not xml_match: chat_history.append( ChatMessage( roleassistant, contentfull_response, ) ) return OutputEvent( responsefull_response, chat_historychat_history, statestate, ) else: xml_str xml_match.group(1) root ET.fromstring(xml_str) plan Plan(steps[]) for step in root.findall(step): plan.steps.append( PlanStep( agent_namestep.attrib[agent], agent_inputstep.text.strip() if step.text else , ) ) return ExecuteEvent(planplan, chat_historychat_history)4.7 执行步骤实现step async def execute(self, ctx: Context, ev: ExecuteEvent) - InputEvent: chat_history ev.chat_history plan ev.plan for step in plan.steps: agent self.agents[step.agent_name] agent_input step.agent_input ctx.write_event_to_stream( PlanEvent( step_infof{step.agent_input} ), ) if step.agent_name ResearchAgent: await call_research_agent(ctx, agent_input) elif step.agent_name WriteAgent: # Note: we arent passing the input from the plan since # were using the state to drive the write agent await call_write_agent(ctx) elif step.agent_name ReviewAgent: await call_review_agent(ctx) state await ctx.store.get(state) chat_history.append( ChatMessage( roleuser, contentfIve completed the previous steps, heres the updated state:\n\n\n{state}\n\n\nDo you need to continue and plan more steps?, If not, write a final response., ) ) return InputEvent( chat_historychat_history, )4.8 运行工作流planner_workflow PlannerWorkflow(timeoutNone) handler planner_workflow.run( 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. ), chat_history[], state{ research_notes: [], report_content: Not written yet., review: Review required., }, ) current_agent None current_tool_calls async for event in handler.stream_events(): if isinstance(event, PlanEvent): print(Executing plan step: , event.step_info) elif isinstance(event, ExecuteEvent): print(Executing plan: , event.plan) result await handler4.9 获取最终报告print(result.response) # 获取最终报告内容 state await handler.ctx.store.get(state) print(state[report_content]) # 获取评审反馈 print(state[review])5. 案例效果通过自定义规划多智能体系统我们成功实现了以下功能自动规划多步骤任务流程将复杂报告生成任务分解为研究、写作和评审三个阶段ResearchAgent能够搜索网络信息并记录结构化的研究笔记WriteAgent能够根据研究笔记生成格式化的Markdown报告ReviewAgent能够评审报告并提供具体的改进建议规划器能够根据评审结果自动规划额外的修订步骤通过流式输出实时展示工作流执行过程状态管理确保各智能体之间的信息传递和持久化工作流程示例步骤1ResearchAgent - 记录关于互联网历史的研究笔记包括互联网发展、万维网出现和21世纪进展↓步骤2WriteAgent - 根据记录的研究笔记起草关于互联网历史的报告包括早期发展、万维网创建和影响以及21世纪里程碑↓步骤3ReviewAgent - 评审生成的报告并提供反馈确保报告符合请求标准↓步骤4WriteAgent - 根据评审建议修订报告包括澄清IETF角色、添加TCP/IP转换日期、解释浏览器战争等最终生成的报告包含了互联网发展的完整历史从ARPANET到现代互联网从万维网的诞生到21世纪的技术进步结构清晰内容详实并包含了权威参考文献。6. 案例实现思路1. 智能体分工设计将报告生成任务分解为三个专门的智能体研究、写作和评审每个智能体负责特定阶段的任务实现职责分离和专业化处理。2. 状态管理机制使用Context.store实现状态持久化确保各智能体之间的信息传递和共享。状态包括研究笔记、报告内容和评审反馈作为智能体之间协作的桥梁。3. 规划器设计创建自定义工作流PlannerWorkflow使用LLM根据用户请求和当前状态生成执行计划。计划以XML格式表示包含有序的步骤列表每个步骤指定要调用的智能体和输入消息。4. 事件驱动架构定义多种事件类型InputEvent、OutputEvent、StreamEvent、PlanEvent、ExecuteEvent实现工作流内部和与外部通信的事件驱动机制支持流式输出和实时状态更新。5. 迭代优化流程通过评审智能体的反馈机制实现报告的迭代优化。当评审不通过时规划器会自动生成新的修订步骤指导写作智能体根据反馈进行改进直到满足质量要求。7. 扩展建议功能扩展添加更多专业智能体如数据分析师、图表生成器实现并行任务执行提高处理效率添加多语言支持实现多语言报告生成集成更多数据源如学术数据库、内部文档实现报告模板系统支持不同格式和风格智能增强实现更智能的规划算法支持条件分支和循环添加智能体间协商机制解决冲突和优化协作实现自动质量评估减少人工评审需求添加学习能力从历史执行中优化规划策略实现用户偏好学习个性化报告风格用户体验添加可视化界面直观展示工作流执行状态实现交互式规划允许用户调整和确认计划添加实时预览功能展示报告生成过程实现版本控制支持报告历史查看和回滚添加协作功能支持多用户共同编辑集成与部署与企业系统集成如CRM、ERP系统实现API接口供其他系统调用添加权限管理和安全控制实现分布式部署提高系统可扩展性添加监控和日志系统便于运维管理8. 总结本案例展示了如何使用LlamaIndex构建一个具有规划能力的多智能体系统实现了从研究到最终报告的完整自动化流程。通过将复杂任务分解为多个专门智能体并使用顶层规划器协调它们的工作我们创建了一个灵活、可扩展的系统架构。案例中的关键技术点包括智能体分工设计、状态管理机制、规划器设计、事件驱动架构和迭代优化流程。这些技术的组合使系统能够自动处理复杂的多步骤任务并根据反馈进行自我优化。这种多智能体规划系统不仅适用于报告生成还可以扩展到其他复杂任务领域如项目管理、产品开发、科研流程等。通过添加更多专业智能体和优化规划算法可以进一步增强系统的能力和适用范围。