Почему ConversationalRetrievalChain стоит рассматривать как legacy-подход, и как мигрировать диалоговый RAG на современные Runnable/agent workflow?

Устаревший паттерн

from langchain.chains import ConversationalRetrievalChain
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.memory import ConversationBufferMemory

vectorstore = Chroma(
    persist_directory="./chroma",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)
retriever = vectorstore.as_retriever()
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    retriever=retriever,
    memory=memory
)
result = chain.invoke({"question": "What is vector search?"})
print(result["answer"])

Что такое ConversationalRetrievalChain и его ограничения

ConversationalRetrievalChain — это устаревший класс LangChain, который объединял историю диалога, конденсацию вопроса (question condensing) и retrieval в один монолитный объект. Внутри него всегда происходили два LLM-вызова: первый — чтобы переформулировать вопрос с учётом истории, второй — чтобы ответить на переформулированный вопрос с найденными документами. Класс помечен deprecated в LangChain 0.2+.

Миграция на LCEL с явной историей

Современный подход: история передаётся явно через MessagesPlaceholder, а переформулировка вопроса — отдельный шаг, который можно контролировать и тестировать независимо:

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

llm = ChatOpenAI(model="gpt-4o-mini")
vectorstore = Chroma(
    persist_directory="./chroma",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Шаг 1: переформулировать вопрос с учётом истории
condense_prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "Given the chat history and a follow-up question, "
        "rewrite the question to be standalone."
    )),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])

condense_chain = condense_prompt | llm | StrOutputParser()

# Шаг 2: ответить на переформулированный вопрос
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer based on the context:\n\n{context}"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])

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

def get_context(inputs: dict) -> dict:
    question = inputs["question"]
    history = inputs.get("chat_history", [])
    if history:
        standalone = condense_chain.invoke({
            "question": question,
            "chat_history": history
        })
    else:
        standalone = question
    docs = retriever.invoke(standalone)
    return {
        "context": format_docs(docs),
        "question": question,
        "chat_history": history
    }

rag_chain = RunnableLambda(get_context) | qa_prompt | llm | StrOutputParser()

# Вызов с историей
chat_history = [
    HumanMessage(content="What is RAG?"),
    AIMessage(content="RAG is Retrieval-Augmented Generation...")
]
result = rag_chain.invoke({
    "question": "How does it differ from fine-tuning?",
    "chat_history": chat_history
})
print(result)

Миграция на LangGraph с персистентной памятью

Если нужна многопользовательская персистентная история между HTTP-запросами, используйте LangGraph с MemorySaver или SqliteSaver:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage
from typing import TypedDict, Sequence

class RAGState(TypedDict):
    messages: Sequence[BaseMessage]
    context: str

def retrieve_node(state: RAGState) -> RAGState:
    last_human = next(
        m for m in reversed(state["messages"])
        if isinstance(m, HumanMessage)
    )
    docs = retriever.invoke(last_human.content)
    state["context"] = format_docs(docs)
    return state

def answer_node(state: RAGState) -> RAGState:
    # используем qa_prompt из примера выше
    response = (qa_prompt | llm | StrOutputParser()).invoke({
        "question": state["messages"][-1].content,
        "context": state["context"],
        "chat_history": state["messages"][:-1]
    })
    state["messages"] = list(state["messages"]) + [AIMessage(content=response)]
    return state

graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve_node)
graph.add_node("answer", answer_node)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "answer")
graph.add_edge("answer", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# thread_id изолирует историю разных пользователей
config = {"configurable": {"thread_id": "user-42"}}
result = app.invoke(
    {"messages": [HumanMessage(content="What is vector search?")], "context": ""},
    config=config
)
print(result["messages"][-1].content)

Подводные камни

• Два LLM-вызова — это два источника стоимости и задержки: condensing-шаг всегда стоит токенов, даже если первый вопрос в сессии не требует переформулировки — добавьте условие if chat_history.
• ConversationBufferMemory не thread-safe: если несколько HTTP-запросов одного пользователя обращаются к одному объекту памяти одновременно, история может перемешаться.
• memory_key должен совпадать с именем переменной в промпте: если в MessagesPlaceholder(variable_name="chat_history") написано одно, а в memory_key другое — цепочка молча потеряет историю.
• return_messages=True обязателен: без него память возвращает строку вместо списка BaseMessage, и MessagesPlaceholder падает с TypeError.
• История растёт неограниченно: без ConversationSummaryMemory или явного trimming после 50+ обменов контекст превысит лимит токенов модели.
• Standalone question может исказить смысл: если condensing-промпт плохо написан, переформулированный вопрос теряет нюансы оригинала — логируйте intermediate шаг в LangSmith.
• Стриминг в ConversationalRetrievalChain ограничен: в LCEL стриминг работает нативно через chain.stream(), но только для последнего LLM-шага, а не для condensing.

• Отвечать определением без production-сценария.
• Не называть runtime boundary, security boundary или failure mode.
• Игнорировать версию API, observability и тестовую проверку.

• Объясняет механизм своими словами и без выдуманных API.
• Называет реальные риски, диагностику и критерий корректности.
• Связывает ответ с текущей документацией и миграционными ограничениями.