本教程旨在指导用户如何在基于langchain的rag（检索增强生成）应用中，利用langserve实现动态输入。我们将重点介绍如何通过langchain表达式语言（lcel）构建可接受动态问题和语言参数的链，并将其部署为langserve服务，从而允许用户在运行时灵活地提供输入，实现交互式问答体验。
引言：Langserve与动态RAG的结合
在构建基于RAG（Retrieval Augmented Generation）的语言模型应用时，通常需要根据用户的实时输入动态地检索相关文档并生成答案。Langserve作为langchain的部署工具，能够将Langchain表达式语言（LCEL）构建的链轻松地暴露为API服务。然而，初学者可能会遇到如何将动态的用户输入（例如查询问题和目标语言）传递给R链的内部组件（如检索器和提示模板）的挑战。本教程将详细阐述如何利用LCEL的强大功能，结合Langserve，构建一个能够处理动态输入的RAG应用。
核心概念：Langchain表达式语言（LCEL）与动态输入
Langchain表达式语言（LCEL）是构建复杂链的基础，它提供了RunnablePassthrough、RunnableParallel和itemgetter等组件，使得处理动态输入变得异常灵活。
- RunnablePassthrough: 允许输入数据直接通过，或在传递前进行简单的转换。
- RunnableParallel: 允许同时运行多个Runnable，并将它们的输出组合成一个字典。这对于为后续步骤准备多个输入非常有用。
- itemgetter: 从字典输入中提取特定键的值。
通过这些组件，我们可以将Langserve接收到的整体输入（通常是一个字典）解构，并将不同的部分传递给链中相应的组件。
构建动态RAG链
我们将构建一个RAG链，它接受一个包含question（用户查询）和lang（目标语言）的字典作为输入。
1. 初始化RAG组件
首先，我们需要一个检索器（Retriever）、一个大型语言模型（LLM）和一个聊天提示模板（ChatpromptTemplate）。
from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.output_parsers import StrOutputParser from langchain_community.vectorstores import FAISS # 示例，实际可替换为其他向量库 from langchain_community.embeddings import openaiEmbeddings # 示例，实际可替换为其他嵌入模型 from langchain_community.document_loaders import TextLoader # 示例 from langchain.text_splitter import CharacterTextSplitter # 示例 from langchain_openai import ChatOpenAI # 示例，实际可替换为其他LLM from operator import itemgetter import os # 假设已经配置了OPENAI_API_KEY环境变量 # os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY" # 1. 模拟数据加载和索引 # 实际应用中，这里会加载你的文档并创建向量存储 # 为了示例，我们创建一个简单的内存向量存储 with open("example_docs.txt", "w") as f: f.write("Finance accounts typically contain records of financial transactions, including income, expenses, assets, liabilities, and equity. They are used to track a company's financial performance and position.n") f.write("A balance sheet provides a snapshot of a company's assets, liabilities, and owner's equity at a specific point in time.n") f.write("The income statement reports a company's financial performance over a specific accounting period, showing revenues and expenses.n") loader = TextLoader("example_docs.txt") documents = loader.load() text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0) docs = text_splitter.split_documents(documents) embeddings = OpenAIEmbeddings() vectorstore = FAISS.from_documents(docs, embeddings) retriever_llm = vectorstore.as_retriever() # 注意：这里命名为retriever_llm是为了与原问题保持一致，实际是retriever # 2. 初始化LLM llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 3. 定义提示模板 # 提示模板将接受 'context', 'question' 和 'lang' 作为输入 template = """根据以下上下文信息回答问题，并严格用{lang}语言输出： {context} 问题: {question} 答案:""" prompt = ChatPromptTemplate.from_template(template) # 辅助函数：格式化检索到的文档 def format_docs(docs): return "nn".join([doc.page_content for doc in docs])
2. 构建LCEL链以处理动态输入
关键在于如何将Langserve接收到的字典输入 ({“question”: “…”, “lang”: “…”}) 分别传递给检索器和提示模板。
# 定义RAG链 rag_chain = ( # 1. 并行处理输入： # - 'context': 从输入中提取 'question'，传递给检索器，然后格式化文档。 # - 'question': 直接从输入中提取 'question'。 # - 'lang': 直接从输入中提取 'lang'。 RunnableParallel({ "context": itemgetter("question") | retriever_llm | format_docs, "question": itemgetter("question"), "lang": itemgetter("lang") }) | prompt # 2. 将处理后的输入传递给提示模板 | llm # 3. 将提示模板的输出传递给LLM | StrOutputParser() # 4. 解析LLM的输出为字符串 )
在这个链中：
- RunnableParallel 允许我们同时执行三个操作，并将结果整合成一个字典。
- itemgetter(“question”) 从原始输入字典中提取question字段。
- 提取出的question被送入retriever_llm进行文档检索。
- 检索到的文档通过format_docs函数进行格式化，作为context。
- 原始的question和lang也分别作为question和lang字段传递。
- 最终，prompt会接收到一个包含context、question和lang的字典，从而能够动态地生成完整的提示。
使用Langserve部署应用
现在，我们将这个动态RAG链部署为Langserve应用。我们需要定义一个Pydantic模型来指定Langserve期望的输入类型。
from fastapi import FastAPI from langserve import add_routes from pydantic import BaseModel, Field # 定义Langserve的输入模型 class InputQuestion(BaseModel): question: str = Field(..., description="The user's query for the RAG system.") lang: str = Field("English", description="The desired output language (e.g., 'English', 'Chinese').") app = FastAPI( title="Dynamic RAG Langserve Application", version="1.0", description="A RAG application with dynamic question and language inputs." ) # 添加路由 # input_type 参数确保Langserve知道如何解析传入的jsON请求体 add_routes( app, rag_chain, path="/dynamic-rag", input_type=InputQuestion, # 指定输入模型 # output_type=str # 如果需要，可以指定输出类型，默认通常是字符串 ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="localhost", port=8000)
运行与测试
-
安装依赖:
pip install langchain langchain-openai langchain-community langserve uvicorn fastapi pydantic
-
运行应用:
python app.py
-
访问Langserve Playground: 打开浏览器访问 http://localhost:8000/dynamic-rag/playground/。 在Playground界面，你将看到一个表单，其中包含 question 和 lang 两个输入字段。你可以动态输入问题和目标语言进行测试。
示例输入:
-
question: “What does finance accounts contain?”
-
lang: “English”
-
question: “财务账目包含什么？”
-
lang: “Chinese”
-
注意事项与总结
- 错误处理: 在生产环境中，应考虑添加更健壮的错误处理机制，例如对LLM调用失败或检索器无结果的情况进行处理。
- 输入验证: Pydantic模型提供了基本的输入验证，但你可以根据需要添加更复杂的验证逻辑。
- 安全性: 对于生产部署，确保API密钥等敏感信息通过环境变量或安全配置管理，而不是硬编码。
- 可扩展性: LCEL链的模块化设计使得你可以轻松地替换或添加新的组件，例如不同的检索器、LLM或后处理步骤。
- configurable_alternate: 虽然本教程主要使用RunnableParallel和itemgetter处理动态输入，但Langchain也提供了configurable_alternate用于更复杂的条件路由场景，例如根据某个配置参数动态选择不同的子链。对于本例中的简单动态参数传递，当前方法更为直接和高效。
通过本教程，你已经掌握了如何利用Langchain表达式语言和Langserve，构建一个能够灵活处理动态输入（如用户查询和目标语言）的RAG应用。这种方法为开发交互式、用户友好的AI应用奠定了坚实的基础。