使用RunnableWithMessageHistory时自定义Context传入对话式RAG链的问题

已通过SelfQueryRetriever获取正确检索结果，并将元数据解析为字符串格式的metadata_context，希望直接传入RAG链以避免重复调用Retriever，但尝试在Prompt模板添加{context}占位符、invoke时传入context参数等操作均无效，链仍使用Retriever返回的原始Document作为Context。

使用create_retrieval_chain构建的RAG链会强制调用传入的Retriever，并将检索结果自动注入到context变量中，手动传入的context参数会被覆盖。同时create_history_aware_retriever也会依赖Retriever生成历史感知的查询，进一步触发重复检索逻辑。

要跳过自动检索逻辑，直接使用自定义的metadata_context，需重构链结构，移除自动检索依赖，改为手动传入上下文：

• 移除历史感知检索器的Retriever绑定，直接构建历史感知查询生成链
• 构建独立的问答链，直接使用传入的metadata_context填充Prompt
• 调整链的组合方式，确保context参数能正确传递到模板中

修改后的核心代码 #

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

def query(query_text, prompt, session_id, metadata_context):
    # 构建历史感知的查询生成Prompt
    contextualize_q_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", contextualize_q_system_prompt),
            ("system", "{context}"),
            ("system", prompt),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ]
    )
    # 直接创建查询生成链，不绑定Retriever
    history_aware_chain = contextualize_q_prompt | llm | StrOutputParser()

    # 构建问答Prompt
    qa_prompt = ChatPromptTemplate.from_messages(
        [   
            ("system", PROMPT_TEMPLATE),
            ("system", "{context}"),
            ("system", prompt),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ]
    )
    # 直接创建问答链
    question_answer_chain = qa_prompt | llm | StrOutputParser()

    # 组合链：生成历史感知查询后传入问答链，保留上下文参数
    rag_chain = RunnablePassthrough.assign(
        # 用历史感知链生成优化后的查询
        input=lambda x: history_aware_chain.invoke({
            "input": x["input"],
            "context": x["context"],
            "chat_history": x["chat_history"]
        }),
        # 传递原始上下文
        context=lambda x: x["context"]
    ) | question_answer_chain

    conversational_rag_chain = RunnableWithMessageHistory(
        rag_chain,
        get_session_history,
        input_messages_key="input",
        history_messages_key="chat_history",
        output_messages_key="answer",
    )
    try:
        logger.info(f"Model: {LLM_MODEL} assigned. Generation of response has started.")
        response = conversational_rag_chain.invoke(
            {"input": query_text, "context": metadata_context},
            config={"configurable": {"session_id": f"{session_id}"}}
        )
        logger.info(f"Response generated.")
    except Exception as e:
        return {'Generation of response failed: ': str(e)}
    return response

关键调整点 #

1. 替换create_history_aware_retriever为手动构建的查询生成链，避免触发Retriever调用
2. 替换create_stuff_documents_chain和create_retrieval_chain为直接的Prompt+LLM链，手动控制上下文注入
3. 使用RunnablePassthrough.assign确保context参数在链传递过程中不丢失
4. 若无需历史感知查询，可直接简化为问答链，跳过历史感知环节

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