Расскажите о случае, когда вы использовали LangGraph для pipeline, evaluation, optimization, deployment или debugging.

Задача: RAG-агент с адаптивным retrieval

Нужно было построить агента для ответов на вопросы по внутренней документации компании. Проблема простого RAG: документация неполная, часть вопросов требовала актуальных данных из интернета. Решение — граф с условным web-поиском.

Архитектура графа

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    question: str
    documents: Annotated[list[str], operator.add]
    generation: str
    web_search_needed: bool
    iterations: int

Узлы графа

from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.tools import TavilySearchResults

vectorstore = Chroma(embedding_function=OpenAIEmbeddings(), persist_directory="./db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
web_search = TavilySearchResults(max_results=3, api_key="tvly-...")
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def retrieve(state: AgentState) -> AgentState:
    docs = retriever.invoke(state["question"])
    return {"documents": [d.page_content for d in docs]}

def grade_documents(state: AgentState) -> str:
    """Проверяем релевантность через LLM."""
    grader_prompt = f"""Are these documents relevant to '{state['question']}'?
    Documents: {state['documents'][:2]}
    Answer yes or no only."""
    result = llm.invoke(grader_prompt).content.lower()
    if "no" in result or len(state["documents"]) == 0:
        return "web_search"
    return "generate"

def web_search_node(state: AgentState) -> AgentState:
    results = web_search.invoke(state["question"])
    web_docs = [r["content"] for r in results]
    return {"documents": web_docs, "web_search_needed": True}

def generate(state: AgentState) -> AgentState:
    context = "\n\n".join(state["documents"][:6])
    prompt = f"""Answer based on context:\n{context}\n\nQuestion: {state['question']}"""
    response = llm.invoke(prompt)
    return {"generation": response.content, "iterations": state.get("iterations", 0) + 1}

Сборка и компиляция

graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve)
graph.add_node("web_search", web_search_node)
graph.add_node("generate", generate)

graph.set_entry_point("retrieve")
graph.add_conditional_edges(
    "retrieve",
    grade_documents,
    {"web_search": "web_search", "generate": "generate"}
)
graph.add_edge("web_search", "generate")
graph.add_edge("generate", END)

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

Запуск и debugging

config = {"configurable": {"thread_id": "session-001"}, "recursion_limit": 5}

# Streaming для отладки промежуточных шагов
for event in app.stream({"question": "What is the deployment process?"}, config=config):
    for key, value in event.items():
        print(f"Node: {key}")
        if "generation" in value:
            print(f"Answer: {value['generation'][:200]}")

Оценка качества

После запуска добавили LangSmith evaluation: собирали вопросы из реального трафика, вручную размечали корректные ответы и запускали evaluate раз в неделю для выявления деградации.

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

• Grader на основе LLM добавляет 500–800ms к каждому запросу и стоит дополнительных токенов — для высоконагруженных систем замените на embedding-similarity score.
• SqliteSaver не thread-safe при параллельных запросах — в production заменили на PostgreSQL через langgraph-checkpoint-postgres.
• Web search через Tavily стоит денег за каждый запрос — без кэширования одинаковых вопросов расходы быстро растут.
• Streaming через app.stream() несовместим с некоторыми ASGI-фреймворками без дополнительной обёртки для Server-Sent Events.
• State schema нельзя изменить без миграции checkpoints — при добавлении нового поля все старые сессии теряют это поле.

• Выдумывать опыт или говорить слишком общими фразами
• Не объяснять свою личную роль в работе с LangGraph
• Не показывать результат, метрики или извлечённые уроки

• Может подготовить честный пример использования LangGraph
• Показывает свою роль, решения и результат
• Умеет рефлексировать над trade-offs и уроками