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

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

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

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

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

Узлы графа

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 и уроками