Создание автономных AI-агентов с LangGraph и DeepSeek-R1: Полная продакшн-архитектура

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    task: str
    plan: List[str]
    current_step: int
    test_results: str
    is_finished: bool

def planning_node(state: AgentState):
    """DeepSeek-R1 формирует план действий."""
    # Имитация вызова модели рассуждения
    print("Глубокое планирование через DeepSeek-R1...")
    return {
        "plan": ["Написать эндпоинт", "Добавить валидацию Pydantic", "Запустить тесты"],
        "current_step": 0
    }

def execution_node(state: AgentState):
    """Исполнение текущего шага плана."""
    step_idx = state["current_step"]
    current_action = state["plan"][step_idx]
    print(f"Исполнение шага: {current_action}")
    return {"current_step": step_idx + 1}

def should_continue(state: AgentState):
    if state["current_step"] >= len(state["plan"]):
        return "finish"
    return "execute"

workflow = StateGraph(AgentState)
workflow.add_node("planner", planning_node)
workflow.add_node("executor", execution_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_conditional_edges("executor", should_continue, {
    "execute": "executor",
    "finish": END
})

# Компиляция с сохранением состояния в памяти
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)