Інтеграція 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 додає абстракцію, яка обґрунтована при складних конвеєрах. Для простого одноразового виклику LLM прямого використання SDK (OpenAI, Anthropic) простіше та передбачуваніше. LangChain оптимальний коли: кілька компонентів (retriever + LLM + parser), кілька провайдерів, потрібні трасування та пам'ять.

Графік реалізації

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