LangChain: От Новичка до Профессионала

1) Введение и дорожная карта

LangChain — это фреймворк для сборки LLM‑приложений из модульных блоков: моделей, промптов, цепочек, памяти, агентов и ретриверов. Вместо «монолитных» скриптов вы строите чётко организованные пайплайны, которые проще тестировать, масштабировать и сопровождать.

Дорожная карта обучения

2) Необходимые основы Python

• Виртуальные окружения: python -m venv .venv → source .venv/bin/activate (Windows: .venv\Scripts\activate)
• Файлы конфигурации: .env, pyproject.toml / requirements.txt
• Асинхронность: async/await, asyncio
• Типизация: typing, TypedDict, Pydantic

# .env (пример)
OPENAI_API_KEY=sk-...
# main.py
import os
from dotenv import load_dotenv
load_dotenv()
print(os.environ.get("OPENAI_API_KEY", "no-key"))

Синхронно vs асинхронно

# Синхронный вызов
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("Скажи привет {name}")
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
print(chain.invoke({"name": "Алекс"}))

# Асинхронный вызов с потоковой отдачей токенов
import asyncio

async def main():
    llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
    stream_chain = prompt | llm | StrOutputParser()
    async for chunk in stream_chain.astream({"name": "миру"}):
        print(chunk, end="")

asyncio.run(main())

Типизация и Pydantic

from typing import List, Optional
from pydantic import BaseModel, Field

class Person(BaseModel):
    name: str = Field(..., description="Имя человека")
    age: int = Field(..., ge=0, le=150)
    email: Optional[str] = Field(None, format="email")
    skills: List[str] = Field(default_factory=list)

3) Установка и настройка окружения

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# Базовые пакеты
pip install -U pip
pip install "langchain>=0.3" "langchain-core>=0.3" "langchain-community>=0.3"
pip install langchain-openai langgraph "langserve[all]" python-dotenv

# Документы и векторные БД
pip install pypdf unstructured docx2txt chromadb faiss-cpu langchain-huggingface

# Для локальных моделей (пример: Ollama)
pip install ollama

Настройка API ключей

import os
from dotenv import load_dotenv

load_dotenv()

# OpenAI
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

# Anthropic
os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY")

# Hugging Face
os.environ["HUGGINGFACEHUB_API_TOKEN"] = os.getenv("HUGGINGFACEHUB_API_TOKEN")

4) Мини‑глоссарий

5) Промпты (Prompt Engineering)

От простой строки к шаблонам

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# PromptTemplate
pt = PromptTemplate.from_template("Объясни {concept} для {audience}")
print(pt.format(concept="градиентный спуск", audience="школьника"))

# ChatPromptTemplate (с ролями)
chat_pt = ChatPromptTemplate.from_messages([
    ("system", "Ты опытный преподаватель ИИ. Отвечай кратко и по делу."),
    ("human", "Разъясни {concept} на примере.")
])

Few‑shot примеры и частичное заполнение

from langchain_core.prompts import FewShotPromptTemplate

examples = [
  {"q": "Что такое переменная?", "a": "Именованная ячейка памяти."},
  {"q": "Что такое функция?", "a": "Переиспользуемый блок кода."}
]

example_prompt = PromptTemplate.from_template("В: {q}\nО: {a}\n")

few = FewShotPromptTemplate(
    examples=examples, 
    example_prompt=example_prompt,
    suffix="В: {q}", 
    input_variables=["q"]
)

partial = pt.partial(audience="джуниора")

Продвинутые техники промптинга

# Промпт с примерами и ограничениями
advanced_prompt = ChatPromptTemplate.from_messages([
    ("system", """Ты эксперт в области программирования.
    Отвечай на русском языке.
    Используй только информацию из контекста.
    Формат ответа: 
    1. Краткое объяснение
    2. Пример кода (если применимо)
    3. Ссылки на дополнительные ресурсы"""),
    ("human", "Объясни {topic}")
])

6) Работа с моделями

Инициализация моделей

from langchain.chat_models import init_chat_model
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.llms import Ollama

# Современный способ (рекомендуется)
openai_model = init_chat_model("gpt-4o-mini", model_provider="openai")
claude_model = init_chat_model("claude-3-haiku", model_provider="anthropic")

# Традиционный способ
openai_direct = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
claude_direct = ChatAnthropic(model="claude-3-haiku", max_tokens=1000)

# Локальные модели
local_model = Ollama(model="llama3")

Параметры моделей

# Основные параметры
model = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,        # Креативность (0-1)
    max_tokens=1000,        # Максимальная длина ответа
    top_p=0.9,              # Разнообразие (nucleus sampling)
    frequency_penalty=0.5,  # Штраф за повторы
    presence_penalty=0.5    # Штраф за новые темы
)

Работа с разными типами моделей

# Чат-модели (рекомендуется для большинства задач)
chat_model = ChatOpenAI(model="gpt-4o-mini")

# Текстовые модели (устаревшие)
from langchain_openai import OpenAI
text_model = OpenAI(model="gpt-3.5-turbo-instruct")

# Embedding модели
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

7) Цепочки и LCEL

LCEL (LangChain Expression Language) позволяет соединять компоненты оператором |.

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")
chain = chat_pt | llm | StrOutputParser()
print(chain.invoke({"concept": "регуляризация"}))

Разветвления и параллель

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

summary_chain = PromptTemplate.from_template("Суммаризуй: {t}") | llm | StrOutputParser()
keywords_chain = PromptTemplate.from_template("Ключевые слова: {t}") | llm | StrOutputParser()

combo = RunnableParallel({
  "summary": summary_chain,
  "keywords": keywords_chain,
  "original": RunnablePassthrough()
})

print(combo.invoke({"t": "Ваш текст"}))

Условные цепочки

from langchain_core.runnables import RunnableLambda

def route_query(query: str) -> str:
    if "математика" in query.lower():
        return "math"
    elif "история" in query.lower():
        return "history"
    else:
        return "general"

def math_chain(input_dict):
    prompt = ChatPromptTemplate.from_template("Реши математическую задачу: {question}")
    return (prompt | llm | StrOutputParser()).invoke(input_dict)

def history_chain(input_dict):
    prompt = ChatPromptTemplate.from_template("Ответь на исторический вопрос: {question}")
    return (prompt | llm | StrOutputParser()).invoke(input_dict)

router = RunnableLambda(route_query)
math_runnable = RunnableLambda(math_chain)
history_runnable = RunnableLambda(history_chain)

conditional_chain = router | {
    "math": math_runnable,
    "history": history_runnable,
    "general": lambda x: "Общие вопросы обрабатываются другим способом"
}

result = conditional_chain.invoke({"question": "Когда закончилась Вторая мировая война?"})
print(result)

8) Память

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate

template = """История:
{history}
Человек: {input}
Кратко ответь:"""

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template(template)
memory = ConversationBufferMemory(return_messages=True)

chain = LLMChain(llm=llm, prompt=prompt, memory=memory, verbose=True)
print(chain.invoke({"input": "Привет! Я Алекс."})["text"])
print(chain.invoke({"input": "Как меня зовут?"})["text"])

Различные типы памяти

# Ограничение по количеству сообщений
from langchain.memory import ConversationBufferWindowMemory
window_memory = ConversationBufferWindowMemory(k=3)  # Только последние 3 сообщения

# Сводка диалога
summary_memory = ConversationSummaryMemory(llm=llm)

# Векторная память для поиска релевантных частей истории
from langchain.memory import VectorStoreRetrieverMemory
# Требует настройки векторной базы данных

# Память с возвратом ключей
from langchain.memory import ConversationBufferMemory
memory_with_keys = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

Использование памяти в LCEL

from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import HumanMessage, AIMessage

# Создание памяти
memory = ConversationBufferMemory(return_messages=True)

# Функция для получения истории
def get_history(inputs):
    return memory.load_memory_variables({})["history"]

# Цепочка с памятью
chain_with_memory = (
    {"history": RunnableLambda(get_history), "input": RunnablePassthrough()}
    | ChatPromptTemplate.from_messages([
        ("system", "Ты полезный ассистент. История диалога: {history}"),
        ("human", "{input}")
    ])
    | llm
    | StrOutputParser()
)

# Сохранение сообщений
memory.save_context({"input": "Привет"}, {"output": "Привет! Как дела?"})

# Вызов цепочки
result = chain_with_memory.invoke("Как меня зовут?")
print(result)

9) Агенты и инструменты

Tool Calling (современный подход)

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

@tool
def calc(expr: str) -> str:
    "Вычисляет простое выражение Python."
    try:
        return str(eval(expr))
    except Exception as e:
        return f"Ошибка: {e}"

tools = [calc]
llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты полезный ассистент. Если нужно посчитать — зови калькулятор."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

print(executor.invoke({"input": "Посчитай 15*4 + 2"})["output"])

Подключение внешнего API (пример GitHub)

import requests
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field

class IssueInput(BaseModel):
    owner: str = Field(..., description="Владелец репо")
    repo: str = Field(..., description="Имя репозитория")

class GitHubIssuesTool(BaseTool):
    name = "github_issues"
    description = "Список последних issues из публичного GitHub репозитория"
    args_schema = IssueInput
    
    def _run(self, owner: str, repo: str):
        url = f"https://api.github.com/repos/{owner}/{repo}/issues"
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        titles = [i["title"] for i in r.json() if "title" in i][:5]
        return "\n".join(titles)

github_tool = GitHubIssuesTool()

Создание кастомного агента

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool

# Создание инструментов
def get_time(query: str) -> str:
    from datetime import datetime
    return f"Текущее время: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

time_tool = Tool(
    name="current_time",
    func=get_time,
    description="Получить текущее время и дату"
)

# Создание агента ReAct
from langchain import hub
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    tools=[time_tool, calc],
    prompt=prompt
)

agent_executor = AgentExecutor(agent=agent, tools=[time_tool, calc], verbose=True)
result = agent_executor.invoke({"input": "Какое сейчас время и сколько будет 10*5?"})
print(result["output"])

10) RAG: Retrieval‑Augmented Generation

1. Загрузка и разбиение документов.
2. Эмбеддинги и векторная БД.
3. Ретривер и генерация ответа c контекстом.

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1) загрузка и разбиение
loader = PyPDFLoader("docs/manual.pdf")
docs = loader.load()
splits = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200).split_documents(docs)

# 2) векторизация
vectorstore = Chroma.from_documents(splits, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

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

# 3) промпт + цепочка
prompt = ChatPromptTemplate.from_template(
"""Ответь, используя только контекст ниже.
Контекст:
{context}
Вопрос: {question}
Краткий ответ:""")

rag = (
  {"context": retriever | format_docs, "question": RunnablePassthrough()}
  | prompt
  | ChatOpenAI(model="gpt-4o-mini")
  | StrOutputParser()
)

print(rag.invoke("О чём документ и какие три основных пункта?"))

Режимы поиска и улучшение качества

# similarity, threshold, MMR
retriever = vectorstore.as_retriever(
    search_type="mmr", 
    search_kwargs={"k": 6, "lambda_mult": 0.3}
)

# Компрессия контекста (уменьшаем шум)
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4o-mini"))
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor, 
    base_retriever=retriever
)

Продвинутый RAG с гибридным поиском

# Гибридный поиск (BM25 + векторы)
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 ретривер
bm25_retriever = BM25Retriever.from_documents(splits)

# Векторный ретривер
vector_retriever = vectorstore.as_retriever()

# Ансамбль ретриверов
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever], 
    weights=[0.5, 0.5]
)

11) LangGraph: сложные пайплайны

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

class State(TypedDict):
    messages: Annotated[Sequence[str], operator.add]

def step1(state: State) -> State: 
    return {"messages": ["Шаг 1"]}

def step2(state: State) -> State: 
    return {"messages": ["Шаг 2"]}

workflow = StateGraph(State)
workflow.add_node("s1", step1)
workflow.add_node("s2", step2)
workflow.set_entry_point("s1")
workflow.add_edge("s1", "s2")
workflow.add_edge("s2", END)

app = workflow.compile()
print(app.invoke({"messages": []}))

Условные переходы и циклы

def router(state: State) -> str:
    return "s2" if len(state["messages"]) < 3 else "end"

workflow.add_conditional_edges(
    "s1", 
    router, 
    {"s2": "s2", "end": END}
)

Пример: Агент с LangGraph

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next: str

def call_model(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    # Логика для определения продолжения
    if len(state["messages"]) > 5:
        return "end"
    return "continue"

# Создание графа
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "agent",
        "end": END
    }
)

app = workflow.compile()

12) Парсинг и валидация вывода

from typing import List
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class Person(BaseModel):
    name: str = Field(..., description="Имя")
    age: int = Field(..., ge=0)
    skills: List[str] = Field(default_factory=list)

parser = JsonOutputParser(pydantic_object=Person)

prompt = ChatPromptTemplate.from_template(
  "Верни JSON по схеме. {format_instructions}\nИмя: Иван, Возраст: 25, Навыки: Python, ML"
).partial(format_instructions=parser.get_format_instructions())

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | parser
print(chain.invoke({}))

Кастомные парсеры

from langchain_core.output_parsers import BaseOutputParser

class CommaSeparatedListOutputParser(BaseOutputParser[List[str]]):
    """Парсит вывод в список строк, разделенных запятыми."""
    
    def parse(self, text: str) -> List[str]:
        return text.strip().split(",")

    def get_format_instructions(self) -> str:
        return "Выведите ответ в виде списка, разделенного запятыми"

# Использование
parser = CommaSeparatedListOutputParser()
chain = prompt | llm | parser
result = chain.invoke({"topic": "основные цвета радуги"})
print(result)

Обработка ошибок парсинга

from langchain_core.runnables import RunnableLambda

def handle_parsing_errors(output):
    try:
        return parser.parse(output)
    except Exception as e:
        # В случае ошибки запросить повторный ответ
        return {"error": "Не удалось распарсить ответ", "raw_output": output}

# Цепочка с обработкой ошибок
safe_chain = prompt | llm | RunnableLambda(handle_parsing_errors)
result = safe_chain.invoke({"input": "..."})

13) Работа с документами

Загрузчики документов

from langchain_community.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    UnstructuredFileLoader,
    WebBaseLoader
)

# PDF
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()

# Word
docx_loader = Docx2txtLoader("document.docx")
docx_docs = docx_loader.load()

# Текст
text_loader = TextLoader("document.txt")
text_docs = text_loader.load()

# Веб-страница
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()

Разделение документов

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)

# Рекурсивное разделение
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", " ", ""]
)

# Разделение по символам
char_splitter = CharacterTextSplitter(
    separator="\n\n",
    chunk_size=1000,
    chunk_overlap=200
)

# Разделение по токенам
token_splitter = TokenTextSplitter(chunk_size=100, chunk_overlap=0)

splits = recursive_splitter.split_documents(docs)

Продвинутая обработка документов

# Извлечение метаданных
from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
docs = loader.load()

# Добавление кастомных метаданных
for doc in docs:
    doc.metadata["source"] = "document.pdf"
    doc.metadata["category"] = "technical"

# Фильтрация документов
filtered_docs = [doc for doc in docs if len(doc.page_content) > 100]

14) Векторные базы данных

Chroma (локальная БД)

from langchain_community.vectorstores import Chroma

# Создание локальной векторной БД
db = Chroma.from_documents(documents, OpenAIEmbeddings(), persist_directory="./chroma_db")
db.persist()  # Сохранение на диск

# Загрузка существующей БД
db = Chroma(persist_directory="./chroma_db", embedding_function=OpenAIEmbeddings())
retriever = db.as_retriever()

FAISS (Facebook AI Similarity Search)

from langchain_community.vectorstores import FAISS

# Создание FAISS индекса
db = FAISS.from_documents(documents, OpenAIEmbeddings())

# Сохранение и загрузка
db.save_local("faiss_index")
db = FAISS.load_local("faiss_index", OpenAIEmbeddings(), allow_dangerous_deserialization=True)

Pinecone (облачная БД)

import os
from langchain_pinecone import PineconeVectorStore
from pinecone import Pinecone

# Инициализация Pinecone
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index_name = "my-langchain-index"

# Создание или подключение к индексу
db = PineconeVectorStore.from_documents(
    documents, 
    OpenAIEmbeddings(), 
    index_name=index_name
)

retriever = db.as_retriever()

Продвинутые операции с векторными БД

# Поиск с фильтрацией
retriever = db.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {"category": "technical"}
    }
)

# Добавление документов
db.add_documents(new_documents)

# Удаление документов
db.delete(ids=["doc1", "doc2"])

# Обновление документов
db.update_documents(updated_documents)

15) Callbacks и отладка

Базовые callbacks

from langchain.callbacks import StdOutCallbackHandler

# Использование callback handler
handler = StdOutCallbackHandler()
result = chain.invoke({"input": "Тест"}, config={"callbacks": [handler]})

Пользовательский callback

from langchain.callbacks.base import BaseCallbackHandler

class TokenCounterCallback(BaseCallbackHandler):
    def __init__(self):
        self.token_count = 0
    
    def on_llm_end(self, response, **kwargs):
        # Подсчет токенов (пример для OpenAI)
        if hasattr(response, 'llm_output') and response.llm_output:
            usage = response.llm_output.get('token_usage', {})
            self.token_count += usage.get('total_tokens', 0)

# Использование
token_counter = TokenCounterCallback()
result = chain.invoke({"input": "Тест"}, config={"callbacks": [token_counter]})
print(f"Использовано токенов: {token_counter.token_count}")

Отладка цепочек

# Включение подробного вывода
chain = prompt | llm | parser
result = chain.invoke({"input": "Тест"}, config={"verbose": True})

# Пошаговая отладка
for step in chain.steps:
    print(f"Шаг: {step}")

# Проверка входных данных
try:
    result = chain.invoke({"input": "Тест"})
    print(result)
except Exception as e:
    print(f"Ошибка: {e}")

16) Тестирование, отладка и мониторинг

Pytest для цепочки

# tests/test_chain.py
import pytest
from myapp.chain import chain

def test_chain_basic():
    out = chain.invoke({"topic": "AI"})
    assert isinstance(out, str)
    assert len(out) > 0

def test_chain_with_memory():
    # Тестирование цепочки с памятью
    result1 = chain.invoke({"input": "Привет"})
    result2 = chain.invoke({"input": "Как дела?"})
    assert "привет" in result1.lower() or "здравств" in result1.lower()

Трассировка LangSmith

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_..."
os.environ["LANGCHAIN_PROJECT"] = "Course Demo"
# Любые вызовы LangChain теперь трассируются в LangSmith

Callbacks и метрики

from langchain.callbacks.base import BaseCallbackHandler

class TokenCounter(BaseCallbackHandler):
    def __init__(self): 
        self.tokens = 0
    
    def on_llm_end(self, response, **kwargs):
        try:
            self.tokens += response.llm_output["token_usage"]["total_tokens"]
        except Exception: 
            pass

counter = TokenCounter()
res = chain.invoke({"topic": "оптимизация"}, config={"callbacks": [counter]})
print("использовано токенов:", counter.tokens)

Тестирование RAG

def test_rag_retrieval():
    """Тестирование корректности поиска документов"""
    query = "Что такое машинное обучение?"
    docs = retriever.get_relevant_documents(query)
    assert len(docs) > 0
    # Проверка релевантности (пример)
    assert any("машинное" in doc.page_content.lower() for doc in docs)

def test_rag_generation():
    """Тестирование генерации ответов"""
    question = "Объясните концепцию нейронных сетей"
    answer = rag_chain.invoke(question)
    assert isinstance(answer, str)
    assert len(answer) > 0
    # Проверка, что ответ содержит ключевые термины
    assert "нейрон" in answer.lower()

17) Деплой: LangServe + FastAPI + Docker

Экспорт цепочки как API

# server.py
from fastapi import FastAPI
from langserve import add_routes
from myapp.chain import rag

app = FastAPI(title="RAG API", version="1.0.0")
add_routes(app, rag, path="/rag")

# запуск:
# uvicorn server:app --host 0.0.0.0 --port 8000

Клиент

from langserve import RemoteRunnable

remote = RemoteRunnable("http://localhost:8000/rag/")
print(remote.invoke({"question": "Что в документе важного?"}))

Dockerfile

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn","server:app","--host","0.0.0.0","--port","8000"]

Best practices деплоя

• Кеширование эмбеддингов и индекса.
• Настройка таймаутов и ограничений на размер входа.
• Обновление индекса во вторичном процессе (не в веб‑воркере).
• Здоровье‑чеки и авто‑рестарт (systemd / Docker healthcheck).

Конфигурация для продакшена

# prod_config.py
import os
from pydantic import BaseSettings

class Settings(BaseSettings):
    openai_api_key: str
    model_name: str = "gpt-4o-mini"
    max_tokens: int = 1000
    temperature: float = 0.7
    vectorstore_path: str = "./vectorstore"
    
    class Config:
        env_file = ".env"

settings = Settings()

18) Практические проекты

📄 A. Чат с PDF

1. Соберите индекс (PDF → чанки → эмбеддинги → Chroma).
2. Поднимите RAG‑цепочку и LangServe.
3. Сделайте простой фронтенд (Streamlit) к API.

# app_streamlit.py (упрощённо)
import streamlit as st
from langserve import RemoteRunnable

st.title("Chat with PDF")
remote = RemoteRunnable("http://localhost:8000/rag/")
q = st.text_input("Ваш вопрос:")
if st.button("Спросить"):
    st.write(remote.invoke({"question": q}))

🤖 B. Агент‑ассистент разработчика

• Инструменты: калькулятор, GitHub, поиск по коду (ripgrep).
• Ограничения: чёткая роль, макс. глубина, whitelisting инструментов.

🧠 C. База знаний компании

• Ингест: Confluence/Google Docs → индексация.
• Контроль качества: MMR, компрессия контекста, тесты на датасетах.

⚙️ D. Автоматизация рабочих процессов

• Интеграция с Jira/Trello для создания задач.
• Агент для автоматического анализа и классификации писем.
• Система для генерации отчетов по данным.

19) FAQ и приёмы оптимизации

20) Полезные ресурсы

• Документация LangChain (Python): python.langchain.com
• LangGraph: langchain-ai.github.io/langgraph
• LangServe: python.langchain.com/docs/langserve/
• LangSmith: docs.smith.langchain.com
• LangChain Academy (бесплатные курсы): academy.langchain.com
• GitHub репозиторий LangChain: github.com/langchain-ai/langchain
• Шаблоны проектов: github.com/langchain-ai/langchain/tree/master/templates

Рекомендуемые книги и статьи

• "Building LLM Applications" - практическое руководство по созданию LLM-приложений
• "Prompt Engineering Guide" - руководство по эффективному промптингу
• "Retrieval-Augmented Generation for NLP" - научная статья о RAG
• "Agents and Tools in LLM Applications" - обзор агентных систем

Сообщества и форумы

• LangChain Discord - активное сообщество разработчиков
• Reddit r/LangChain - обсуждение новостей и вопросов
• Stack Overflow - вопросы и ответы по LangChain
• GitHub Issues - официальная поддержка и баг-репорты

Подсказка: закрепите версии пакетов и сохраняйте артефакты индекса. Это снижает сюрпризы в проде.