Imported from Alex1980Alex/1C-Framework (.claude/skills/langchain-core/SKILL.md). Install upstream with npx skills add Alex1980Alex/1C-Framework --skill langchain-core. Copyright stays with the author.

LangChain Core

Установка

pip install -U langchain
pip install -U langchain-anthropic   # или langchain-openai, langchain-google-genai

Python 3.10+ обязателен. Провайдеры ставятся отдельными пакетами.

create_agent — точка входа

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

model = init_chat_model("claude-sonnet-4-5-20250929", temperature=0.7, max_tokens=1000, timeout=30)

agent = create_agent(
    model=model,
    tools=[get_weather, search_db],
    system_prompt="You are a helpful assistant",
    context_schema=Context,                    # Опционально: типизация контекста
    response_format=ToolStrategy(ResponseFormat), # Опционально: структурированный вывод
    checkpointer=InMemorySaver(),              # Опционально: память
    middleware=[LoggingMiddleware()],           # Опционально: middleware
    state_schema=CustomAgentState              # Опционально: расширение состояния
)

config = {"configurable": {"thread_id": "1"}}
result = agent.invoke({"messages": [{"role": "user", "content": "Hello"}]}, config)

Инструменты (@tool)

Базовый инструмент

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search customer database for records matching query.
    Args:
        query: Search query
        limit: Max results
    """
    return f"Found {limit} results for '{query}'"

ToolRuntime — доступ к контексту, store, stream

from langchain.tools import tool, ToolRuntime

@tool
def fetch_user_info(runtime: ToolRuntime[Context]) -> str:
    """Get user info from context and store."""
    user_id = runtime.context.user_id        # Static context
    if runtime.store:                         # Cross-session memory
        if mem := runtime.store.get(("users",), user_id):
            return mem.value["preferences"]
    return "Default preferences"

Обновление состояния агента из tool

from langgraph.types import Command
from langchain.messages import RemoveMessage, REMOVE_ALL_MESSAGES

@tool
def update_user_name(new_name: str, runtime: ToolRuntime) -> Command:
    """Update user name in state."""
    return Command(update={"user_name": new_name})

@tool
def clear_conversation() -> Command:
    """Clear conversation history."""
    return Command(update={"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]})

stream_writer в инструментах

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    writer = runtime.stream_writer
    writer(f"Fetching data for: {city}")
    return f"In {city} it's sunny!"

Модели

Инициализация

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    temperature=0.7,    # 0 = deterministic, 1+ = creative
    max_tokens=1000,
    timeout=30,
    max_retries=3
)

invoke / stream / batch

response = model.invoke("Question?")                     # Single
for chunk in model.stream("Question?"):                   # Streaming
    print(chunk.text, end="", flush=True)
responses = model.batch(["Q1", "Q2", "Q3"])               # Batch
for r in model.batch_as_completed(inputs, config={"max_concurrency": 5}):  # Parallel batch
    print(r)

Tool calling

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("Weather in Boston?")
for tc in response.tool_calls:
    print(f"Tool: {tc['name']}, Args: {tc['args']}")

Structured output на уровне модели

class Movie(BaseModel):
    title: str = Field(..., description="Movie title")
    year: int
    director: str

model_structured = model.with_structured_output(Movie)
movie = model_structured.invoke("Tell me about Inception")
# Movie(title="Inception", year=2010, director="Nolan")

Сообщения

from langchain.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage

messages = [
    SystemMessage("You are a poetry expert"),
    HumanMessage("Write haiku about spring"),
    AIMessage("Blossoms fall softly..."),
    ToolMessage(content="Result", tool_call_id="call_123", name="search")
]

Мультимодальный контент

# Изображение
HumanMessage(content=[
    {"type": "text", "text": "Describe this image"},
    {"type": "image", "url": "https://example.com/image.jpg"}
])

# PDF
HumanMessage(content=[
    {"type": "text", "text": "Analyze document"},
    {"type": "file", "base64": "...", "mime_type": "application/pdf"}
])

Токены

response.usage_metadata  # {'input_tokens': 8, 'output_tokens': 304, 'total_tokens': 312}

Structured Output (на уровне агента)

ProviderStrategy (нативная поддержка модели)

from langchain.agents.structured_output import ProviderStrategy

agent = create_agent(model="gpt-4.1", response_format=ProviderStrategy(ContactInfo))
result = agent.invoke({"messages": [...]})
print(result["structured_response"])  # ContactInfo(...)

ToolStrategy (через tool calling — более универсальная)

from langchain.agents.structured_output import ToolStrategy

agent = create_agent(model="gpt-4.1", response_format=ToolStrategy(ProductReview))
# Поддержка Union типов:
response_format=ToolStrategy(Union[ContactInfo, EventDetails])
# Обработка ошибок:
response_format=ToolStrategy(schema=ProductReview, handle_errors=True)

Middleware

Встроенные middleware

Custom middleware (декоратор)

from langchain.agents.middleware import before_model, after_model, wrap_model_call

@before_model
def log_before(state: AgentState, runtime: Runtime) -> dict | None:
    print(f"Messages: {len(state['messages'])}")
    return None  # None = no state change

@wrap_model_call
def retry_model(request: ModelRequest, handler: Callable) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            if attempt == 2: raise

@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(content=f"Error: {e}", tool_call_id=request.tool_call["id"])

Dynamic model/tool selection через middleware

@wrap_model_call
def dynamic_model(request: ModelRequest, handler) -> ModelResponse:
    model = advanced_model if len(request.state["messages"]) > 10 else basic_model
    return handler(request.override(model=model))

@wrap_model_call
def filter_tools(request: ModelRequest, handler) -> ModelResponse:
    role = request.runtime.context.user_role
    tools = request.tools if role == "admin" else [t for t in request.tools if "read" in t.name]
    return handler(request.override(tools=tools))

Runtime — контекст, состояние, store

Dynamic prompts

from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def prompt_from_context(request: ModelRequest) -> str:
    return f"You are helpful. User: {request.runtime.context.user_name}"

Guardrails (PII)

PIIMiddleware("email", strategy="redact", apply_to_input=True)
PIIMiddleware("credit_card", strategy="mask", detector=r"\d{4}-\d{4}-\d{4}-\d{4}")

# Custom detector
def detect_ssn(content: str) -> list[dict]:
    return [{"text": m.group(0), "start": m.start(), "end": m.end()}
            for m in re.finditer(r"\d{3}-\d{2}-\d{4}", content)]

PIIMiddleware("ssn", detector=detect_ssn, strategy="hash")

Стратегии: block | redact | mask | hash

Human-in-the-Loop

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="gpt-4.1",
    tools=[read_email, send_email],
    checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"send_email_tool": {"allowed_decisions": ["approve", "edit", "reject"]},
                      "read_email_tool": False}
    )]
)

# 1. Запуск (прерывается на send_email_tool)
result1 = agent.invoke({"messages": [...]}, config)

# 2. Возобновление с решением
result2 = agent.invoke({
    "messages": result1["messages"],
    "decisions": [{"tool_call_id": "call_123", "decision": "approve"}]
}, config)

Решения: approve | edit (с новыми args) | reject (с причиной)

RAG паттерны

2-Step RAG

docs = retriever.invoke(query)
response = model.invoke(f"Answer based on: {docs}\n\nQ: {query}")

Agentic RAG (агент решает когда искать)

@tool
def fetch_documentation(query: str) -> str:
    """Fetch docs for query."""
    return "\n".join([d.page_content for d in retriever.invoke(query)])

agent = create_agent(model="gpt-4.1", tools=[fetch_documentation])

Источники: Lang Chain Docs/Lang Chain/ — 18 файлов (Обзор, Установка, Быстрый старт, Агенты, Инструменты, Модели, Сообщения, Кратковременная память, Структурированный вывод, Middleware (3), Ограждения, Среда выполнения, Контекстная инженерия, Извлечение, Человек в процессе, Философия)