企业官网建设流程全解析

案例目标

本案例展示了如何使用LlamaIndex工作流实现路由查询引擎(Router Query Engine)。路由查询引擎能够根据查询内容自动选择最合适的查询引擎来处理用户查询，从多个查询选项中选择最合适的工具来回答问题。这种设计模式特别适用于处理多种类型的查询，如摘要查询、特定上下文查询和关键词查询等。

技术栈与核心依赖

• llama-index- 核心框架，提供工作流和查询引擎功能
• llama-index-llms-openai- OpenAI语言模型集成
• llama-index-core- LlamaIndex核心组件
• OpenAI GPT-4o-mini- 用于查询路由选择的LLM
• TreeSummarize- 用于合并多个查询引擎响应的摘要器

环境配置

# 安装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: SelectorResult

2. 路由查询引擎工作流实现

创建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_multi=select_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_engines=selected_query_engines ) @step async def generate_responses(self, ctx: Context, ev: QueryEngineSelectionEvent) -> SynthesizeEvent: """从选定的查询引擎生成响应""" query = await ctx.store.get("query", default=None) 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(f"number 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(f"Selected 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(f"Selected query engine: {selected_query_engines.ind}: {selected_query_engines.reason}") response_generated = [await query_engine.aquery(query)] return SynthesizeEvent( result=response_generated, selected_query_engines=selected_query_engines, ) @step async def synthesize_responses(self, ctx: Context, ev: SynthesizeEvent) -> StopEvent: """合成为最终响应""" response_generated = ev.result query = await ctx.store.get("query", default=None) 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_str=query) ) else: response = response_generated[0] response.metadata = response.metadata or {} response.metadata["selector_result"] = selected_query_engines return StopEvent(result=response)

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_mode="tree_summarize", use_async=True, ) vector_query_engine = vector_index.as_query_engine() keyword_query_engine = keyword_index.as_query_engine() # 创建工具 list_tool = QueryEngineTool.from_defaults( query_engine=list_query_engine, description=( "Useful for summarization questions related to Paul Graham eassy on" " What I Worked On." ), ) vector_tool = QueryEngineTool.from_defaults( query_engine=vector_query_engine, description=( "Useful for retrieving specific context from Paul Graham essay on What" " I Worked On." ), ) keyword_tool = QueryEngineTool.from_defaults( query_engine=keyword_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(model="gpt-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( llm=llm, summary_template=DEFAULT_TREE_SUMMARIZE_PROMPT_SEL, )

6. 运行工作流

创建工作流实例并运行查询：

import nest_asyncio nest_asyncio.apply() # 创建工作流实例 w = RouterQueryEngineWorkflow(timeout=200) # 示例1: 摘要查询 query = "Provide the summary of the document?" result = await w.run( query=query, llm=llm, query_engine_tools=query_engine_tools, summarizer=summarizer, select_multi=True, # 可以选择多个查询引擎 ) # 示例2: 特定上下文查询 query = "What did the author do growing up?" result = await w.run( query=query, llm=llm, query_engine_tools=query_engine_tools, summarizer=summarizer, select_multi=False, # 只选择一个查询引擎 ) # 示例3: 可能需要多个查询引擎的查询 query = "What were noteable events and people from the authors time at Interleaf and YC?" result = await w.run( query=query, llm=llm, query_engine_tools=query_engine_tools, summarizer=summarizer, select_multi=True, # 启用多查询引擎选择 )

案例效果

本案例实现的路由查询引擎能够根据查询内容自动选择最合适的查询引擎：

• 摘要查询：对于"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"的信息，表明正在进行响应合并。

案例实现思路

1. 事件驱动架构：使用LlamaIndex工作流的事件驱动模型，通过定义QueryEngineSelectionEvent和SynthesizeEvent等事件类，实现步骤间的数据传递。
2. 智能路由选择：利用LLM的语义理解能力，根据查询内容自动选择最合适的查询引擎。支持单选和多选两种模式。
3. 多引擎并行处理：当选择多个查询引擎时，使用asyncio.gather实现并行查询，提高处理效率。
4. 响应合并机制：使用TreeSummarize将多个查询引擎的响应合并为统一的答案，保持答案的一致性和完整性。
5. 元数据保留：在最终响应中保留选择器结果(selected_query_engines)作为元数据，便于后续分析和调试。

扩展建议

• 动态查询引擎注册：实现运行时动态添加和移除查询引擎的能力，使系统更加灵活。
• 查询引擎性能监控：添加性能指标收集，记录每个查询引擎的选择频率和响应质量，用于优化选择策略。
• 自定义选择器：实现基于查询历史、用户偏好或特定领域的自定义选择器，提高路由准确性。
• 多模态查询支持：扩展支持图像、音频等多模态查询，为每种模态创建专门的查询引擎。
• 查询结果缓存：实现查询结果缓存机制，对相同或相似查询直接返回缓存结果，提高响应速度。
• 用户反馈集成：添加用户反馈机制，根据用户对查询结果的满意度调整选择策略。

总结

路由查询引擎工作流是一个强大的查询处理模式，它通过智能选择最合适的查询引擎来优化查询效果。本案例展示了如何使用LlamaIndex工作流实现这一模式，包括事件定义、工作流步骤设计、多查询引擎集成和响应合并等关键技术。

这种设计特别适用于处理多种类型的查询场景，如文档问答系统、知识库检索和复合信息查询等。通过将不同类型的查询分配给专门优化的查询引擎，系统能够提供更准确、更相关的答案，同时保持良好的用户体验。

该实现的核心优势在于其灵活性和可扩展性，开发者可以轻松添加新的查询引擎类型，调整选择策略，或集成更复杂的响应合并逻辑，从而构建适应特定应用场景的智能查询系统。