案例目标本案例展示了如何使用LlamaIndex工作流实现路由查询引擎(Router Query Engine)。路由查询引擎能够根据查询内容自动选择最合适的查询引擎来处理用户查询从多个查询选项中选择最合适的工具来回答问题。这种设计模式特别适用于处理多种类型的查询如摘要查询、特定上下文查询和关键词查询等。技术栈与核心依赖llama-index- 核心框架提供工作流和查询引擎功能llama-index-llms-openai- OpenAI语言模型集成llama-index-core- LlamaIndex核心组件OpenAI GPT-4o-mini- 用于查询路由选择的LLMTreeSummarize- 用于合并多个查询引擎响应的摘要器环境配置# 安装LlamaIndex pip install -U llama-index # 设置OpenAI API密钥 import os os.environ[OPENAI_API_KEY] sk-..案例实现1. 定义事件类首先定义工作流中使用的事件类用于在不同步骤之间传递数据from llama_index.core.workflow import Event from llama_index.core.base.base_selector import SelectorResult from typing import Dict, List, Any from llama_index.core.base.response.schema import RESPONSE_TYPE class QueryEngineSelectionEvent(Event): 选择查询引擎工具的结果 selected_query_engines: SelectorResult class SynthesizeEvent(Event): 从不同查询引擎合成响应的事件 result: List[RESPONSE_TYPE] selected_query_engines: SelectorResult2. 路由查询引擎工作流实现创建RouterQueryEngineWorkflow类包含三个主要步骤class RouterQueryEngineWorkflow(Workflow): step async def selector(self, ctx: Context, ev: StartEvent) - QueryEngineSelectionEvent: 根据查询选择单个/多个查询引擎 # 获取输入参数 await ctx.store.set(query, ev.get(query)) await ctx.store.set(llm, ev.get(llm)) await ctx.store.set(query_engine_tools, ev.get(query_engine_tools)) await ctx.store.set(summarizer, ev.get(summarizer)) # 获取LLM和选择器 llm Settings.llm select_multiple_query_engines ev.get(select_multi) query ev.get(query) query_engine_tools ev.get(query_engine_tools) selector get_selector_from_llm(llm, is_multiselect_multiple_query_engines) # 获取查询引擎元数据 query_engines_metadata [ query_engine.metadata for query_engine in query_engine_tools ] # 选择最合适的查询引擎 selected_query_engines await selector.aselect( query_engines_metadata, query ) return QueryEngineSelectionEvent( selected_query_enginesselected_query_engines ) step async def generate_responses(self, ctx: Context, ev: QueryEngineSelectionEvent) - SynthesizeEvent: 从选定的查询引擎生成响应 query await ctx.store.get(query, defaultNone) selected_query_engines ev.selected_query_engines query_engine_tools await ctx.store.get(query_engine_tools) query_engines [engine.query_engine for engine in query_engine_tools] print(fnumber of selected query engines: {len(selected_query_engines.selections)}) if len(selected_query_engines.selections) 1: # 并行处理多个查询引擎 tasks [] for selected_query_engine in selected_query_engines.selections: print(fSelected query engine: {selected_query_engine.index}: {selected_query_engine.reason}) query_engine query_engines[selected_query_engine.index] tasks.append(query_engine.aquery(query)) response_generated await asyncio.gather(*tasks) else: # 处理单个查询引擎 query_engine query_engines[selected_query_engines.selections[0].index] print(fSelected query engine: {selected_query_engines.ind}: {selected_query_engines.reason}) response_generated [await query_engine.aquery(query)] return SynthesizeEvent( resultresponse_generated, selected_query_enginesselected_query_engines, ) step async def synthesize_responses(self, ctx: Context, ev: SynthesizeEvent) - StopEvent: 合成为最终响应 response_generated ev.result query await ctx.store.get(query, defaultNone) summarizer await ctx.store.get(summarizer) selected_query_engines ev.selected_query_engines if len(response_generated) 1: # 合并多个查询引擎的响应 response await self.acombine_responses( summarizer, response_generated, QueryBundle(query_strquery) ) else: response response_generated[0] response.metadata response.metadata or {} response.metadata[selector_result] selected_query_engines return StopEvent(resultresponse)3. 数据准备和索引创建加载文档数据并创建多种类型的索引# 下载文档数据 !mkdir -p data/paul_graham/ !wget https://raw.githubusercontent.com/run-llama/llama_index/main/docs/examples/data/paul_graham/paul_graham_essay.txt -O data/paul_graham/paul_graham_essay.txt # 加载数据 from llama_index.core import SimpleDirectoryReader documents SimpleDirectoryReader(./data/paul_graham).load_data() # 创建节点 nodes Settings.node_parser.get_nodes_from_documents(documents) # 创建三种不同类型的索引 from llama_index.core import ( VectorStoreIndex, SummaryIndex, SimpleKeywordTableIndex, ) summary_index SummaryIndex(nodes) # 摘要索引 vector_index VectorStoreIndex(nodes) # 向量索引 keyword_index SimpleKeywordTableIndex(nodes) # 关键词表索引4. 创建查询引擎工具为每种索引创建查询引擎工具并定义其用途描述from llama_index.core.tools import QueryEngineTool # 创建查询引擎 list_query_engine summary_index.as_query_engine( response_modetree_summarize, use_asyncTrue, ) vector_query_engine vector_index.as_query_engine() keyword_query_engine keyword_index.as_query_engine() # 创建工具 list_tool QueryEngineTool.from_defaults( query_enginelist_query_engine, description( Useful for summarization questions related to Paul Graham eassy on What I Worked On. ), ) vector_tool QueryEngineTool.from_defaults( query_enginevector_query_engine, description( Useful for retrieving specific context from Paul Graham essay on What I Worked On. ), ) keyword_tool QueryEngineTool.from_defaults( query_enginekeyword_query_engine, description( Useful for retrieving specific context using keywords from Paul Graham essay on What I Worked On. ), ) query_engine_tools [list_tool, vector_tool, keyword_tool]5. 配置LLM和摘要器# 配置OpenAI LLM from llama_index.llms.openai import OpenAI from llama_index.core import Settings llm OpenAI(modelgpt-4o-mini) Settings.llm llm # 配置摘要器 from llama_index.core.prompts.default_prompt_selectors import ( DEFAULT_TREE_SUMMARIZE_PROMPT_SEL, ) from llama_index.core.response_synthesizers import TreeSummarize summarizer TreeSummarize( llmllm, summary_templateDEFAULT_TREE_SUMMARIZE_PROMPT_SEL, )6. 运行工作流创建工作流实例并运行查询import nest_asyncio nest_asyncio.apply() # 创建工作流实例 w RouterQueryEngineWorkflow(timeout200) # 示例1: 摘要查询 query Provide the summary of the document? result await w.run( queryquery, llmllm, query_engine_toolsquery_engine_tools, summarizersummarizer, select_multiTrue, # 可以选择多个查询引擎 ) # 示例2: 特定上下文查询 query What did the author do growing up? result await w.run( queryquery, llmllm, query_engine_toolsquery_engine_tools, summarizersummarizer, select_multiFalse, # 只选择一个查询引擎 ) # 示例3: 可能需要多个查询引擎的查询 query What were noteable events and people from the authors time at Interleaf and YC? result await w.run( queryquery, llmllm, query_engine_toolsquery_engine_tools, summarizersummarizer, select_multiTrue, # 启用多查询引擎选择 )案例效果本案例实现的路由查询引擎能够根据查询内容自动选择最合适的查询引擎摘要查询对于Provide the summary of the document?这类查询系统自动选择了摘要查询引擎(list_tool)并生成了文档的全面摘要。特定上下文查询对于What did the author do growing up?这类查询系统选择了向量查询引擎(vector_tool)能够检索到与作者成长经历相关的具体内容。复合查询对于What were noteable events and people from the authors time at Interleaf and YC?这类查询系统同时选择了向量查询引擎和关键词查询引擎并将两者的响应合并提供了更全面的答案。每种查询都显示了选择的查询引擎索引和选择原因使整个查询过程透明化。当选择多个查询引擎时系统还会显示Combining responses from multiple query engines的信息表明正在进行响应合并。案例实现思路事件驱动架构使用LlamaIndex工作流的事件驱动模型通过定义QueryEngineSelectionEvent和SynthesizeEvent等事件类实现步骤间的数据传递。智能路由选择利用LLM的语义理解能力根据查询内容自动选择最合适的查询引擎。支持单选和多选两种模式。多引擎并行处理当选择多个查询引擎时使用asyncio.gather实现并行查询提高处理效率。响应合并机制使用TreeSummarize将多个查询引擎的响应合并为统一的答案保持答案的一致性和完整性。元数据保留在最终响应中保留选择器结果(selected_query_engines)作为元数据便于后续分析和调试。扩展建议动态查询引擎注册实现运行时动态添加和移除查询引擎的能力使系统更加灵活。查询引擎性能监控添加性能指标收集记录每个查询引擎的选择频率和响应质量用于优化选择策略。自定义选择器实现基于查询历史、用户偏好或特定领域的自定义选择器提高路由准确性。多模态查询支持扩展支持图像、音频等多模态查询为每种模态创建专门的查询引擎。查询结果缓存实现查询结果缓存机制对相同或相似查询直接返回缓存结果提高响应速度。用户反馈集成添加用户反馈机制根据用户对查询结果的满意度调整选择策略。总结路由查询引擎工作流是一个强大的查询处理模式它通过智能选择最合适的查询引擎来优化查询效果。本案例展示了如何使用LlamaIndex工作流实现这一模式包括事件定义、工作流步骤设计、多查询引擎集成和响应合并等关键技术。这种设计特别适用于处理多种类型的查询场景如文档问答系统、知识库检索和复合信息查询等。通过将不同类型的查询分配给专门优化的查询引擎系统能够提供更准确、更相关的答案同时保持良好的用户体验。该实现的核心优势在于其灵活性和可扩展性开发者可以轻松添加新的查询引擎类型调整选择策略或集成更复杂的响应合并逻辑从而构建适应特定应用场景的智能查询系统。