案例目标本案例展示了如何使用LlamaIndex Workflow构建一个带重排序功能的RAG检索增强生成系统。该系统通过工作流的方式组织RAG流程包括文档索引、检索、重排序和响应合成四个主要步骤实现了更精确、更高效的信息检索和生成。核心目标构建一个结构化的RAG工作流通过重排序技术提高检索结果的相关性并利用流式响应提供更好的用户体验。技术栈与核心依赖LlamaIndex Workflow- 提供工作流框架和事件驱动架构OpenAI GPT-4o-mini- 用于重排序和响应合成OpenAI Embedding- 用于文档向量化text-embedding-3-small模型VectorStoreIndex- 用于创建和存储文档向量索引LLMRerank- 用于基于LLM的重排序CompactAndRefine- 用于响应合成SimpleDirectoryReader- 用于文档加载OpenInference- 可选用于工作流可观测性环境配置本案例需要配置以下API密钥import os os.environ[OPENAI_API_KEY] sk-proj-...此外需要安装以下依赖pip install -U llama-index pip install openinference-instrumentation-llama-index3.0.0 opentelemetry-proto1.12.0 opentelemetry-exporter-otlp opentelemetry-sdk可选地可以配置Phoenix进行工作流追踪from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPSpanExporter from openinference.instrumentation.llama_index import LlamaIndexInstrumentor # 添加Phoenix API密钥进行追踪 PHOENIX_API_KEY YOUR-PHOENIX-API-KEY os.environ[OTEL_EXPORTER_OTLP_HEADERS] fapi_key{PHOENIX_API_KEY} # 添加Phoenix span_phoenix_processor SimpleSpanProcessor( HTTPSpanExporter(endpointhttps://app.phoenix.arize.com/v1/traces) ) # 添加到追踪器 tracer_provider trace_sdk.TracerProvider() tracer_provider.add_span_processor(span_processorspan_phoenix_processor) # 装备应用程序 LlamaIndexInstrumentor().instrument(tracer_providertracer_provider)案例实现1. 数据准备首先创建数据目录并下载示例文档!mkdir -p data !wget --user-agent Mozilla https://arxiv.org/pdf/2307.09288.pdf -O data/llama2.pdf2. 定义工作流事件定义工作流中使用的自定义事件类from llama_index.core.workflow import Event from llama_index.core.schema import NodeWithScore class RetrieverEvent(Event): 检索运行的结果 nodes: list[NodeWithScore] class RerankEvent(Event): 对检索节点进行重排序的结果 nodes: list[NodeWithScore]3. 定义RAG工作流类创建RAG工作流类包含文档索引、检索、重排序和响应合成四个主要步骤from llama_index.core import SimpleDirectoryReader, VectorStoreIndex from llama_index.core.response_synthesizers import CompactAndRefine from llama_index.core.postprocessor.llm_rerank import LLMRerank from llama_index.core.workflow import ( Context, Workflow, StartEvent, StopEvent, step, ) from llama_index.llms.openai import OpenAI from llama_index.embeddings.openai import OpenAIEmbedding class RAGWorkflow(Workflow): step async def ingest(self, ctx: Context, ev: StartEvent) - StopEvent | None: 文档摄取入口点由带有dirname的StartEvent触发。 dirname ev.get(dirname) if not dirname: return None documents SimpleDirectoryReader(dirname).load_data() index VectorStoreIndex.from_documents( documentsdocuments, embed_modelOpenAIEmbedding(model_nametext-embedding-3-small), ) return StopEvent(resultindex) step async def retrieve( self, ctx: Context, ev: StartEvent ) - RetrieverEvent | None: RAG入口点由带有query的StartEvent触发。 query ev.get(query) index ev.get(index) if not query: return None print(fQuery the database with: {query}) # 将查询存储在全局上下文中 await ctx.store.set(query, query) # 从全局上下文获取索引 if index is None: print(Index is empty, load some documents before querying!) return None retriever index.as_retriever(similarity_top_k2) nodes await retriever.aretrieve(query) print(fRetrieved {len(nodes)} nodes.) return RetrieverEvent(nodesnodes) step async def rerank(self, ctx: Context, ev: RetrieverEvent) - RerankEvent: # 对节点进行重排序 ranker LLMRerank( choice_batch_size5, top_n3, llmOpenAI(modelgpt-4o-mini) ) print(await ctx.store.get(query, defaultNone), flushTrue) new_nodes ranker.postprocess_nodes( ev.nodes, query_strawait ctx.store.get(query, defaultNone) ) print(fReranked nodes to {len(new_nodes)}) return RerankEvent(nodesnew_nodes) step async def synthesize(self, ctx: Context, ev: RerankEvent) - StopEvent: 使用重排序后的节点返回流式响应。 llm OpenAI(modelgpt-4o-mini) summarizer CompactAndRefine(llmllm, streamingTrue, verboseTrue) query await ctx.store.get(query, defaultNone) response await summarizer.asynthesize(query, nodesev.nodes) return StopEvent(resultresponse)4. 运行工作流创建工作流实例并执行文档索引和查询# 创建工作流实例 w RAGWorkflow() # 摄取文档 index await w.run(dirnamedata) # 运行查询 result await w.run(queryHow was Llama2 trained?, indexindex) # 流式输出响应 async for chunk in result.async_response_gen(): print(chunk, end, flushTrue)案例效果本案例实现了一个完整的RAG工作流能够处理文档索引、检索、重排序和响应合成。以下是查询How was Llama2 trained?的执行过程和结果执行过程Query the database with: How was Llama2 trained? Retrieved 2 nodes. How was Llama2 trained? Reranked nodes to 2最终响应Llama 2 was trained through a multi-step process that began with pretraining using publicly available online sources. This was followed by the creation of an initial version of Llama 2-Chat through supervised fine-tuning. The model was then iteratively refined using Reinforcement Learning with Human Feedback (RLHF) methodologies, which included rejection sampling and Proximal Policy Optimization (PPO). During pretraining, the model utilized an optimized auto-regressive transformer architecture, incorporating robust data cleaning, updated data mixes, and training on a significantly larger dataset of 2 trillion tokens. The training process also involved increased context length and the use of grouped-query attention (GQA) to enhance inference scalability. The training employed the AdamW optimizer with specific hyperparameters, including a cosine learning rate schedule and gradient clipping. The models were pretrained on Metas Research SuperCluster and internal production clusters, utilizing NVIDIA A100 GPUs.效果亮点系统通过工作流方式组织RAG流程实现了模块化设计通过重排序技术提高了检索结果的相关性通过流式响应提供了更好的用户体验。案例实现思路工作流程图文档摄取 → 检索 → 重排序 → 响应合成StartEvent(dirname) → StopEvent(index)StartEvent(query, index) → RetrieverEvent(nodes) → RerankEvent(nodes) → StopEvent(response)核心设计思路事件驱动架构使用LlamaIndex Workflow的事件驱动模型通过定义不同类型的事件来控制工作流的执行流程。模块化设计将RAG流程分解为独立的步骤每个步骤负责特定的功能提高了代码的可维护性和可扩展性。双入口点设计工作流有两个入口点一个用于文档索引一个用于查询处理使系统更加灵活。上下文共享利用ctx.store在工作流步骤之间共享状态信息如查询字符串。流式响应使用CompactAndRefine的流式功能提供更好的用户体验。关键技术点类型注解验证工作流自动使用类型注解进行验证确保事件在步骤之间正确传递。异步处理所有步骤都是异步的提高了系统的并发处理能力。重排序技术使用LLMRerank对检索结果进行重排序提高结果相关性。可观测性可选集成OpenInference和Phoenix提供工作流执行的可视化追踪。扩展建议增加多模态支持扩展系统以支持图像、表格等多模态数据的检索和生成。实现查询转换在检索步骤前添加查询转换逻辑提高检索准确性。添加缓存机制为检索和重排序结果添加缓存提高响应速度。增强错误处理添加更健壮的错误处理机制当某个步骤失败时能够自动重试或寻找替代方案。支持多种重排序方法集成多种重排序算法根据查询类型动态选择最适合的方法。添加评估指标集成RAG评估指标如答案相关性、上下文相关性等持续优化系统性能。实现混合检索结合关键词检索和向量检索提高检索的全面性。支持多文档源扩展系统以支持从多种文档源如数据库、API、网页等检索信息。总结本案例展示了如何使用LlamaIndex Workflow构建一个带重排序功能的RAG系统。通过工作流的方式组织RAG流程实现了模块化、可扩展的架构使每个步骤都可以独立优化和替换。重排序技术的引入显著提高了检索结果的相关性而流式响应则提供了更好的用户体验。该案例的核心价值在于提供了一种结构化的RAG实现方法使流程更加清晰和可维护展示了事件驱动架构在RAG系统中的应用演示了如何通过重排序技术提高检索质量实现了流式响应提升了用户体验提供了可选的可观测性功能便于系统监控和调试这种基于工作流的RAG实现方式为构建更复杂、更智能的检索增强生成系统提供了有力参考特别适合需要高度定制化和可扩展性的应用场景。
案例目标本案例展示了如何使用LlamaIndex Workflow构建一个带重排序功能的RAG检索增强生成系统。该系统通过工作流的方式组织RAG流程包括文档索引、检索、重排序和响应合成四个主要步骤实现了更精确、更高效的信息检索和生成。核心目标构建一个结构化的RAG工作流通过重排序技术提高检索结果的相关性并利用流式响应提供更好的用户体验。技术栈与核心依赖LlamaIndex Workflow- 提供工作流框架和事件驱动架构OpenAI GPT-4o-mini- 用于重排序和响应合成OpenAI Embedding- 用于文档向量化text-embedding-3-small模型VectorStoreIndex- 用于创建和存储文档向量索引LLMRerank- 用于基于LLM的重排序CompactAndRefine- 用于响应合成SimpleDirectoryReader- 用于文档加载OpenInference- 可选用于工作流可观测性环境配置本案例需要配置以下API密钥import os os.environ[OPENAI_API_KEY] sk-proj-...此外需要安装以下依赖pip install -U llama-index pip install openinference-instrumentation-llama-index3.0.0 opentelemetry-proto1.12.0 opentelemetry-exporter-otlp opentelemetry-sdk可选地可以配置Phoenix进行工作流追踪from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPSpanExporter from openinference.instrumentation.llama_index import LlamaIndexInstrumentor # 添加Phoenix API密钥进行追踪 PHOENIX_API_KEY YOUR-PHOENIX-API-KEY os.environ[OTEL_EXPORTER_OTLP_HEADERS] fapi_key{PHOENIX_API_KEY} # 添加Phoenix span_phoenix_processor SimpleSpanProcessor( HTTPSpanExporter(endpointhttps://app.phoenix.arize.com/v1/traces) ) # 添加到追踪器 tracer_provider trace_sdk.TracerProvider() tracer_provider.add_span_processor(span_processorspan_phoenix_processor) # 装备应用程序 LlamaIndexInstrumentor().instrument(tracer_providertracer_provider)案例实现1. 数据准备首先创建数据目录并下载示例文档!mkdir -p data !wget --user-agent Mozilla https://arxiv.org/pdf/2307.09288.pdf -O data/llama2.pdf2. 定义工作流事件定义工作流中使用的自定义事件类from llama_index.core.workflow import Event from llama_index.core.schema import NodeWithScore class RetrieverEvent(Event): 检索运行的结果 nodes: list[NodeWithScore] class RerankEvent(Event): 对检索节点进行重排序的结果 nodes: list[NodeWithScore]3. 定义RAG工作流类创建RAG工作流类包含文档索引、检索、重排序和响应合成四个主要步骤from llama_index.core import SimpleDirectoryReader, VectorStoreIndex from llama_index.core.response_synthesizers import CompactAndRefine from llama_index.core.postprocessor.llm_rerank import LLMRerank from llama_index.core.workflow import ( Context, Workflow, StartEvent, StopEvent, step, ) from llama_index.llms.openai import OpenAI from llama_index.embeddings.openai import OpenAIEmbedding class RAGWorkflow(Workflow): step async def ingest(self, ctx: Context, ev: StartEvent) - StopEvent | None: 文档摄取入口点由带有dirname的StartEvent触发。 dirname ev.get(dirname) if not dirname: return None documents SimpleDirectoryReader(dirname).load_data() index VectorStoreIndex.from_documents( documentsdocuments, embed_modelOpenAIEmbedding(model_nametext-embedding-3-small), ) return StopEvent(resultindex) step async def retrieve( self, ctx: Context, ev: StartEvent ) - RetrieverEvent | None: RAG入口点由带有query的StartEvent触发。 query ev.get(query) index ev.get(index) if not query: return None print(fQuery the database with: {query}) # 将查询存储在全局上下文中 await ctx.store.set(query, query) # 从全局上下文获取索引 if index is None: print(Index is empty, load some documents before querying!) return None retriever index.as_retriever(similarity_top_k2) nodes await retriever.aretrieve(query) print(fRetrieved {len(nodes)} nodes.) return RetrieverEvent(nodesnodes) step async def rerank(self, ctx: Context, ev: RetrieverEvent) - RerankEvent: # 对节点进行重排序 ranker LLMRerank( choice_batch_size5, top_n3, llmOpenAI(modelgpt-4o-mini) ) print(await ctx.store.get(query, defaultNone), flushTrue) new_nodes ranker.postprocess_nodes( ev.nodes, query_strawait ctx.store.get(query, defaultNone) ) print(fReranked nodes to {len(new_nodes)}) return RerankEvent(nodesnew_nodes) step async def synthesize(self, ctx: Context, ev: RerankEvent) - StopEvent: 使用重排序后的节点返回流式响应。 llm OpenAI(modelgpt-4o-mini) summarizer CompactAndRefine(llmllm, streamingTrue, verboseTrue) query await ctx.store.get(query, defaultNone) response await summarizer.asynthesize(query, nodesev.nodes) return StopEvent(resultresponse)4. 运行工作流创建工作流实例并执行文档索引和查询# 创建工作流实例 w RAGWorkflow() # 摄取文档 index await w.run(dirnamedata) # 运行查询 result await w.run(queryHow was Llama2 trained?, indexindex) # 流式输出响应 async for chunk in result.async_response_gen(): print(chunk, end, flushTrue)案例效果本案例实现了一个完整的RAG工作流能够处理文档索引、检索、重排序和响应合成。以下是查询How was Llama2 trained?的执行过程和结果执行过程Query the database with: How was Llama2 trained? Retrieved 2 nodes. How was Llama2 trained? Reranked nodes to 2最终响应Llama 2 was trained through a multi-step process that began with pretraining using publicly available online sources. This was followed by the creation of an initial version of Llama 2-Chat through supervised fine-tuning. The model was then iteratively refined using Reinforcement Learning with Human Feedback (RLHF) methodologies, which included rejection sampling and Proximal Policy Optimization (PPO). During pretraining, the model utilized an optimized auto-regressive transformer architecture, incorporating robust data cleaning, updated data mixes, and training on a significantly larger dataset of 2 trillion tokens. The training process also involved increased context length and the use of grouped-query attention (GQA) to enhance inference scalability. The training employed the AdamW optimizer with specific hyperparameters, including a cosine learning rate schedule and gradient clipping. The models were pretrained on Metas Research SuperCluster and internal production clusters, utilizing NVIDIA A100 GPUs.效果亮点系统通过工作流方式组织RAG流程实现了模块化设计通过重排序技术提高了检索结果的相关性通过流式响应提供了更好的用户体验。案例实现思路工作流程图文档摄取 → 检索 → 重排序 → 响应合成StartEvent(dirname) → StopEvent(index)StartEvent(query, index) → RetrieverEvent(nodes) → RerankEvent(nodes) → StopEvent(response)核心设计思路事件驱动架构使用LlamaIndex Workflow的事件驱动模型通过定义不同类型的事件来控制工作流的执行流程。模块化设计将RAG流程分解为独立的步骤每个步骤负责特定的功能提高了代码的可维护性和可扩展性。双入口点设计工作流有两个入口点一个用于文档索引一个用于查询处理使系统更加灵活。上下文共享利用ctx.store在工作流步骤之间共享状态信息如查询字符串。流式响应使用CompactAndRefine的流式功能提供更好的用户体验。关键技术点类型注解验证工作流自动使用类型注解进行验证确保事件在步骤之间正确传递。异步处理所有步骤都是异步的提高了系统的并发处理能力。重排序技术使用LLMRerank对检索结果进行重排序提高结果相关性。可观测性可选集成OpenInference和Phoenix提供工作流执行的可视化追踪。扩展建议增加多模态支持扩展系统以支持图像、表格等多模态数据的检索和生成。实现查询转换在检索步骤前添加查询转换逻辑提高检索准确性。添加缓存机制为检索和重排序结果添加缓存提高响应速度。增强错误处理添加更健壮的错误处理机制当某个步骤失败时能够自动重试或寻找替代方案。支持多种重排序方法集成多种重排序算法根据查询类型动态选择最适合的方法。添加评估指标集成RAG评估指标如答案相关性、上下文相关性等持续优化系统性能。实现混合检索结合关键词检索和向量检索提高检索的全面性。支持多文档源扩展系统以支持从多种文档源如数据库、API、网页等检索信息。总结本案例展示了如何使用LlamaIndex Workflow构建一个带重排序功能的RAG系统。通过工作流的方式组织RAG流程实现了模块化、可扩展的架构使每个步骤都可以独立优化和替换。重排序技术的引入显著提高了检索结果的相关性而流式响应则提供了更好的用户体验。该案例的核心价值在于提供了一种结构化的RAG实现方法使流程更加清晰和可维护展示了事件驱动架构在RAG系统中的应用演示了如何通过重排序技术提高检索质量实现了流式响应提升了用户体验提供了可选的可观测性功能便于系统监控和调试这种基于工作流的RAG实现方式为构建更复杂、更智能的检索增强生成系统提供了有力参考特别适合需要高度定制化和可扩展性的应用场景。
