LangChain+FastAPI开发API：聊天历史与来源返回融合问题求助

ConversationRetrievalChain完全支持自定义RAG模板、保留聊天记忆同时返回来源文档，以下是基于你现有代码的修改版本：

from fastapi import FastAPI
from langchain.vectorstores import FAISS
from langchain.chat_models import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.embeddings import OpenAIEmbeddings
import pandas as pd
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationRetrievalChain
from langchain_core.messages import HumanMessage, AIMessage
from pydantic import BaseModel

model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
embeddings = OpenAIEmbeddings()

vectorstore = FAISS.load_local("products_metadata", embeddings, allow_dangerous_deserialization=True)
retriever = vectorstore.as_retriever()

products = pd.read_json('./data/products.json', orient='records', lines=True)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="Spin up a simple api server using Langchain's Runnable interfaces",
)

# 调整模板，加入聊天历史变量
ANSWER_TEMPLATE ="""使用以下聊天历史和上下文信息回答问题：
聊天历史：{chat_history}
上下文：{context}
问题：{question}

请按照要求回答：开头用引导句，上下文产品信息格式为：- 产品名称，品牌，国家，每个产品换行。
回答："""

ANSWER_PROMPT = ChatPromptTemplate.from_template(ANSWER_TEMPLATE)

# 保留原有的文档格式化与回答生成逻辑
rag_chain_from_docs = (
    RunnablePassthrough.assign(context=(lambda x: format_docs(x["context"])))
    | ANSWER_PROMPT
    | model
    | StrOutputParser()
)

# 配置聊天记忆组件
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

# 构建融合记忆与来源返回的对话RAG链
conversation_rag_chain = ConversationRetrievalChain.from_llm(
    llm=model,
    retriever=retriever,
    memory=memory,
    combine_docs_chain=rag_chain_from_docs,  # 接入自定义的回答生成链
    return_source_documents=True,  # 开启来源文档返回
    verbose=False
)

# 更新请求模型，支持传入聊天历史
class ChatRequest(BaseModel):
    question: str
    chat_history: list[dict] = []  # 格式示例：[{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]

@app.post("/chat")
def chat(request: ChatRequest):
    # 将传入的聊天历史转换为LangChain兼容的消息格式
    history_messages = []
    for msg in request.chat_history:
        if msg["role"] == "user":
            history_messages.append(HumanMessage(content=msg["content"]))
        elif msg["role"] == "assistant":
            history_messages.append(AIMessage(content=msg["content"]))
    
    # 更新记忆存储
    memory.chat_memory.messages = history_messages
    
    # 调用对话链
    result = conversation_rag_chain.invoke({"question": request.question})
    
    # 提取来源文档内容
    source_docs = [doc.page_content for doc in result["source_documents"]]
    
    return {
        "answer": result["answer"],
        "source_documents": source_docs,
        "chat_history": request.chat_history + [{"role": "assistant", "content": result["answer"]}]
    }

@app.get("/products")
def get_products():
    return products.to_dict(orient='records')

@app.get("/products/{product_id}")
def get_product(product_id: int):
    product = products[products['code'] == product_id]
    return product.to_dict(orient='records')

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

关键修改说明 #

1. Prompt适配：在模板中加入{chat_history}变量，让模型能参考历史对话生成连贯回答
2. 记忆组件：使用ConversationBufferMemory存储聊天历史，指定output_key确保与自定义链的输出兼容
3. 链整合：将你原有的RAG回答链作为combine_docs_chain传入ConversationRetrievalChain，保留自定义的文档格式化与回答规则
4. 来源返回：开启return_source_documents=True，链会自动返回检索到的原始文档
5. 接口更新：修改请求模型支持传入聊天历史，转换为LangChain兼容格式后更新记忆，最终返回包含回答、来源和更新后历史的结果

额外建议 #

• 生产环境建议使用RedisChatMessageHistory等持久化记忆存储，避免单进程内存限制
• 长对话场景可替换为ConversationSummaryMemory，压缩历史内容减少token消耗
• 来源文档可根据需求提取更多元数据（如文档ID、存储路径等）

内容的提问来源于stack exchange，提问作者FeckNeck