Интеграция LangChain для AI-пайплайнов

Интеграция LangChain для AI-пайплайнов

LangChain — фреймворк для построения LLM-приложений, предоставляющий абстракции над моделями, документами, инструментами и памятью. Ключевая концепция — цепочки (chains), объединяющие компоненты в воспроизводимые пайплайны обработки. LCEL (LangChain Expression Language) — декларативный интерфейс, унифицирующий синтаксис для любой комбинации компонентов.

LCEL: основа современного LangChain

LCEL использует оператор | для композиции компонентов. Любой объект, реализующий Runnable, можно соединить в цепочку:

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_community.vectorstores import Chroma

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Простая цепочка
prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты — эксперт по {domain}."),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()
result = chain.invoke({"domain": "финансовый анализ", "question": "Что такое EBITDA?"})

# Параллельная цепочка
parallel_chain = RunnableParallel({
    "summary": prompt | llm | StrOutputParser(),
    "keywords": ChatPromptTemplate.from_template("Извлеки ключевые слова: {question}") | llm | StrOutputParser(),
})

Интеграция с LLM-провайдерами

LangChain поддерживает единый интерфейс для разных провайдеров:

# OpenAI
from langchain_openai import ChatOpenAI
llm_openai = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

# Anthropic
from langchain_anthropic import ChatAnthropic
llm_claude = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# Google
from langchain_google_genai import ChatGoogleGenerativeAI
llm_gemini = ChatGoogleGenerativeAI(model="gemini-2.0-flash")

# Локальная Ollama
from langchain_ollama import ChatOllama
llm_local = ChatOllama(model="llama3.2:3b", temperature=0)

# Hugging Face
from langchain_huggingface import HuggingFaceEndpoint
llm_hf = HuggingFaceEndpoint(repo_id="mistralai/Mistral-7B-Instruct-v0.3")

Смена провайдера не требует изменения логики цепочки — только замена объекта llm.

RAG-пайплайн с LangChain

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
from langchain_core.runnables import RunnablePassthrough
import json

# Загрузка и разбивка документов
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
chunks = splitter.split_documents(docs)

# Индексирование
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = QdrantVectorStore.from_documents(
    chunks,
    embedding=embeddings,
    url="http://localhost:6333",
    collection_name="knowledge_base",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# RAG-цепочка
rag_prompt = ChatPromptTemplate.from_template("""Ответь на вопрос на основе контекста.

Контекст:
{context}

Вопрос: {question}

Если ответа нет в контексте — скажи об этом явно.""")

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

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("Какие условия расторжения договора?")

Управление памятью в диалоге

from langchain.memory import ConversationBufferWindowMemory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory

# История чата в Redis
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    return RedisChatMessageHistory(session_id, url="redis://localhost:6379")

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты — ассистент технической поддержки."),
    ("placeholder", "{history}"),
    ("human", "{input}"),
])

chain_with_history = RunnableWithMessageHistory(
    chat_prompt | llm | StrOutputParser(),
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# Использование
config = {"configurable": {"session_id": "user_123"}}
chain_with_history.invoke({"input": "Моё приложение не запускается"}, config=config)
chain_with_history.invoke({"input": "Ошибка: 'connection refused'"}, config=config)

Structured Output и валидация

from pydantic import BaseModel, Field
from typing import Literal

class TicketClassification(BaseModel):
    category: Literal["billing", "technical", "account", "other"]
    priority: Literal["low", "medium", "high", "critical"]
    summary: str = Field(description="Краткое описание проблемы (1-2 предложения)")
    requires_human: bool

structured_llm = llm.with_structured_output(TicketClassification)

classify_prompt = ChatPromptTemplate.from_messages([
    ("system", "Классифицируй обращение в техподдержку."),
    ("human", "{ticket_text}"),
])

classifier_chain = classify_prompt | structured_llm
result: TicketClassification = classifier_chain.invoke({
    "ticket_text": "Я не могу войти в систему, пароль не подходит уже второй день"
})

LangSmith: трассировка и отладка

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_PROJECT"] = "production-rag"

# Все вызовы автоматически логируются в LangSmith
# Доступны: входные/выходные данные, latency, токены, стоимость, трейс

LangSmith позволяет анализировать промпты, находить узкие места в цепочках и сравнивать версии пайплайнов через dataset-evaluation.

Практический кейс: унификация 5 LLM-интеграций

Ситуация: продуктовая команда поддерживала 5 отдельных интеграций (OpenAI, Claude, корпоративный YandexGPT, локальный Llama, Gemini) с дублирующимся кодом retry-логики, форматирования промптов и обработки ошибок.

Решение: рефакторинг на LangChain LCEL с единым интерфейсом.

Архитектура:

• Конфигурируемый провайдер через env-переменную LLM_PROVIDER
• Общие промпт-темплейты в YAML-файлах
• Единый слой обработки ошибок через .with_fallbacks()

from langchain_core.runnables import RunnableWithFallbacks

primary_llm = ChatOpenAI(model="gpt-4o")
fallback_llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

robust_llm = primary_llm.with_fallbacks([fallback_llm])

Результаты:

• Объём кода интеграций: -67%
• Время добавления нового провайдера: 3 дня → 4 часа
• Uptime пайплайна (за счёт fallback): 99.1% → 99.8%
• Видимость в LangSmith: время отладки инцидентов сократилось с 2ч до 20мин

Когда LangChain избыточен

LangChain добавляет абстракцию, которая оправдана при сложных пайплайнах. Для простого one-shot вызова LLM прямое использование SDK (OpenAI, Anthropic) проще и предсказуемее. LangChain оптимален когда: несколько компонентов (retriever + LLM + parser), несколько провайдеров, нужны трассировка и память.

Сроки

• Базовая интеграция LangChain + 1 провайдер: 2–4 дня
• RAG-пайплайн с векторной БД: 1–2 недели
• Диалоговый агент с памятью: 1–2 недели
• Рефакторинг существующего кода на LCEL: 1–3 недели