Langserve中实现动态RAG应用：Langchain链式输入处理教程

pip install langchain langchain-openai langserve uvicorn faiss-cpu

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()
)

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)