本教程详细阐述如何在langserve中构建支持动态输入的rag（检索增强生成）应用。文章通过langchain的runnable接口，展示如何将用户查询和目标语言作为动态参数传递给检索器和llm提示模板，从而实现灵活、可配置的交互式ai服务。内容涵盖链式组件的构建、langserve路由配置及示例代码，帮助开发者轻松部署动态rag解决方案。
在构建基于Langchain的RAG（Retrieval Augmented Generation）应用时，一个常见的需求是能够动态地接收用户输入，例如查询问题和目标语言，而不是将它们硬编码在代码中。当使用Langserve部署这些应用时，实现动态输入是提升应用灵活性和用户体验的关键。本教程将指导您如何通过Langchain的Runnable接口，构建一个能够接受动态输入的Langserve RAG应用。
理解Langserve与动态输入
Langserve允许您将Langchain的Runnable对象作为API端点暴露。当一个Runnable被添加到Langserve应用中时，Langserve会自动解析其输入签名，并在Playground界面中生成相应的输入字段。要实现动态输入，核心在于构建一个Langchain链，使其能够从初始输入中提取所需的参数，并将其传递给链中的各个组件（如检索器、提示模板）。
构建动态RAG链的核心组件
为了实现动态RAG，我们需要将检索、文档格式化、提示构建和LLM调用等步骤串联起来，并确保每个步骤都能正确接收其所需的动态输入。
- 检索器（Retriever）: 检索器需要一个查询字符串作为输入。在动态链中，这个查询字符串将来自用户的初始输入。
- 文档格式化（Document Formatter）: 检索器返回的是文档对象列表，通常需要将其格式化为单一字符串以便填充到提示模板中。
- 提示模板（Prompt Template）: 提示模板需要上下文信息（格式化的文档）和用户问题，以及目标语言。
- 语言模型（LLM）: LLM接收填充好的提示，并生成最终答案。
我们将使用RunnablePassthrough和RunnableLambda来编排这些组件。
逐步构建动态RAG链
首先，确保您已经安装了必要的库：
pip install langchain langchain-openai langserve uvicorn faiss-cpu
接下来，我们将定义一个完整的Langchain RAG链。
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.promnpts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.llms import OpenAI # 示例LLM
from langchain_community.embeddings import OpenAIEmbeddings # 示例嵌入模型
from langchain_community.vectorstores import FAISS # 示例向量存储
from langchain_core.documents import Document
import os
# 确保设置了OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
# 1. 初始化一个示例检索器
# 在实际应用中，这里会加载您的文档并构建一个向量存储
# 为了演示，我们创建一个简单的FAISS向量存储和检索器
documents = [
Document(page_content="财务账户通常包含资产、负债、权益、收入和支出。"),
Document(page_content="主要的财务报表包括利润表、资产负债表和现金流量表。"),
Document(page_content="公司的盈利能力可以通过利润表来评估。"),
Document(page_content="资产负债表反映了公司在特定时间点的财务状况。")
]
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(documents, embeddings)
retriever = vectorstore.as_retriever()
# 2. 初始化LLM
llm = OpenAI(temperature=0) # 可以替换为其他LLM，如ChatOpenAI
# 3. 定义文档格式化函数
def format_docs(docs):
"""将检索到的文档列表格式化为单个字符串"""
return "\n\n".join(doc.page_content for doc in docs)
# 4. 构建动态RAG链
# 链的输入将是一个字典，例如 {"question": "...", "lang": "..."}
rag_chain = (
# 步骤1: 接收初始输入，并使用RunnablePassthrough.assign将检索结果添加到输入字典中
# RunnableLambda(retriever.get_relevant_documents) 包装了检索器方法，
# .bind(input=lambda x: x["question"]) 确保检索器接收到输入字典中的"question"字段
RunnablePassthrough.assign(
documents=RunnableLambda(retriever.get_relevant_documents).bind(input=lambda x: x["question"])
)
# 步骤2: 再次使用RunnablePassthrough.assign，将格式化后的文档作为"context"添加到输入字典中
| RunnablePassthrough.assign(
context=lambda x: format_docs(x["documents"])
)
# 步骤3: 构建ChatPromptTemplate，它会从输入字典中获取"context"、"question"和"lang"
| ChatPromptTemplate.from_template(
"根据以下上下文信息:\n{context}\n\n请回答问题: {question}\n严格用{lang}语言回答。"
)
# 步骤4: 调用LLM
| llm
# 步骤5: 解析LLM的输出为字符串
| StrOutputParser()
)在这个链中：
- RunnablePassthrough.assign(documents=...) 允许我们将检索器的结果（documents）添加到链的输入字典中，供后续步骤使用。RunnableLambda用于将一个普通函数（或方法）包装成一个Runnable。.bind(input=lambda x: x["question"]) 是关键，它告诉RunnableLambda从当前的输入字典x中取出"question"作为检索器的输入。
- 第二个 RunnablePassthrough.assign(context=...) 类似地，将格式化后的文档作为context添加到输入字典。
- ChatPromptTemplate.from_template(...) 会自动从其接收到的输入字典中查找{context}、{question}和{lang}对应的键值来填充模板。
Langserve部署
现在，我们将这个动态RAG链部署到Langserve应用中。
from fastapi import FastAPI
from langserve import add_routes
import uvicorn
# 创建FastAPI应用实例
app = FastAPI(
title="动态RAG应用",
version="1.0",
description="一个支持动态查询和语言的Langserve RAG应用。",
)
# 将RAG链添加到Langserve路由
# input_type参数可以帮助Langserve Playground更好地展示输入字段
add_routes(
app,
rag_chain,
path="/dynamic_rag",
input_type={"question": str, "lang": str},
# 可以在此处添加config_schema来进一步定义可配置项，
# 但对于简单的动态输入，input_type已足够
)
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=8000)运行上述代码，然后访问 http://localhost:8000/dynamic_rag/playground。您会看到Langserve Playground自动生成了question和lang两个输入字段，允许您动态输入查询和目标语言。
注意事项与扩展
- 错误处理与输入验证: 在实际生产环境中，您应该为输入添加更健壮的验证机制，例如使用Pydantic模型来定义输入结构。
- 异步操作: 对于高性能需求，可以考虑使用异步版本的LLM和检索器，并相应地调整链的构建。
- configurable_alternate 的应用: 虽然本教程主要通过标准链式组合实现动态输入，但Langchain的configurable_alternate在更复杂的场景下非常有用。例如，如果您想根据用户输入或配置动态地切换不同的检索器（如，对于财务问题使用财务检索器，对于技术问题使用技术检索器），那么configurable_alternate将是理想的选择。它允许您基于一个配置键来选择执行哪个Runnable。
- 安全性: 部署到生产环境时，务必考虑API密钥的安全性，不要直接硬编码在代码中，而应使用环境变量或秘密管理服务。
- 可观测性: 结合Langsmith等工具可以更好地监控和调试您的Langserve应用。
通过以上步骤，您已经成功构建了一个能够接收动态输入并由Langserve托管的RAG应用。这种方法提供了极大的灵活性，使得您的AI应用能够根据用户的具体需求提供定制化的响应。
