本教程旨在指导用户如何在基于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 "\n\n".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)运行与测试
保存代码: 将上述所有代码保存为一个Python文件，例如app.py。
网页制作与PHP语言应用下载图书《网页制作与PHP语言应用》，由武汉大学出版社于2006出版，该书为普通高等院校网络传播系列教材之一，主要阐述了网页制作的基础知识与实践，以及PHP语言在网络传播中的应用。该书内容涉及：HTML基础知识、PHP的基本语法、PHP程序中的常用函数、数据库软件MySQL的基本操作、网页加密和身份验证、动态生成图像、MySQL与多媒体素材库的建设等。
-
安装依赖:
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应用奠定了坚实的基础。
