Construye un flujo de trabajo de programación tipo Ultracode con LangGraph

Resumen

• LangGraph es el runtime durable del flujo de trabajo principal y la fuente de verdad. Los ciclos de herramientas de los workers usan create_agent de LangChain, que a su vez está construido sobre LangGraph; el diseño no depende de Agno.
• Las llamadas al modelo son nodos normales del grafo con salida estructurada y tipada. LangGraph administra el estado, el fan-out dinámico de workers, el fan-in, los reintentos, las rondas de reparación, los checkpoints, las interrupciones y la reanudación.
• Usa Send para lanzar un worker para una tarea rutinaria o varios workers independientes a partir de un plan generado por el modelo.
• Usa estado con reducers solo para eventos de workers y revisiones que se agregan sin modificar los anteriores. El código determinista, no un modelo reductor, debe construir el manifiesto del candidato y el informe de pruebas.
• Los revisores reciben el SHA exacto del candidato, un diff inmutable y los artefactos de prueba sin procesar. La aprobación queda vinculada a ese mismo candidato antes de su promoción.
• PostgreSQL hace durable el estado del grafo; no proporciona una cola de ejecución ni un sandbox. Agrega un runtime de API/cola y un servicio de espacios de trabajo aislados.
• En producción, garantiza la idempotencia porque LangGraph reanuda desde checkpoints y puede volver a ejecutar nodos.
• Esta aproximación ejecuta una ola de workers independientes por ronda. No intenta ejecutar un DAG de dependencias arbitrario.

Qué aprenderás aquí

• Cómo modelar un ciclo tipo Ultracode como un StateGraph
• Cómo generar workers dinámicos con Send
• Cómo funcionan los reducers y el fan-in
• Cómo preservar criterios de éxito y comandos de regresión inmutables
• Cómo integrar, probar, revisar, reparar y aprobar un candidato
• Cómo usar checkpoints de PostgreSQL y Command(resume=...)
• Cómo desplegar con FastAPI personalizado o Agent Server
• Qué infraestructura queda fuera de LangGraph

Esta es la Parte 3 de la serie Arquitectura de Ultracode. La Parte 1 deriva el diseño portable. La Parte 2 presenta una alternativa independiente con Agno/AgentOS. Este artículo asume que elegiste el ecosistema de LangGraph y no lo combina con Agno.

Por qué LangGraph encaja con este problema

El trabajo tipo Ultracode se parece más a una máquina de estados que a un ciclo de chat:

request
  -> typed plan
  -> dynamic worker wave
  -> candidate integration
  -> deterministic tests
  -> synthesis
  -> verifier + refuter
  -> repair or approval
  -> idempotent promotion

LangGraph proporciona:

Lo que LangGraph no proporciona es igual de importante:

• Worktrees de Git
• aislamiento de procesos
• política de comandos
• almacenamiento de artefactos
• identidad autenticada de quien aprueba
• una cola de ejecución durable en la biblioteca OSS
• cuotas por tenant y facturación

Estos siguen siendo contratos de infraestructura.

Arquitectura

API / queue runtime
       |
       v
LangGraph thread
       |
       +--> plan node
       |
       +--> Send(worker A)
       +--> Send(worker B)
       +--> Send(worker C)
                |
                v
       deterministic integrate node
                |
                v
       deterministic test node
                |
                v
          synthesis node
           /          \
          v            v
     verifier       refuter
           \          /
            review barrier
               |
       fail --> repair plan
               |
              pass
               |
        interrupt approval
               |
        promote candidate SHA

Alcance de la estructura base

El código no se puede ejecutar hasta que proporciones las implementaciones de los adaptadores. El grafo muestra las poscondiciones que deben cumplir y dónde se conectan.

Estructura del proyecto

langgraph-ultracode/
  app/
    schemas.py
    model.py
    workspace.py
    graph.py
    api.py
    bootstrap.py        # Concrete workspace and approval adapters
  tests/
    test_plan.py
    test_resume.py
    test_candidate_binding.py
  pyproject.toml

Las firmas de la API se revisaron el July 15, 2026 frente a Python 3.10+, LangGraph 1.2.9, LangChain 1.3.13, LangChain OpenAI 1.3.5 y LangGraph Checkpoint PostgreSQL 3.1.0.

uv add "langgraph==1.2.9" "langgraph-checkpoint-postgres==3.1.0" \
  "langchain==1.3.13" "langchain-openai==1.3.5" "psycopg[binary,pool]" \
  pydantic fastapi uvicorn

Las API de timeout por nodo y para Send requieren LangGraph 1.2 o posterior.

Paso 1: Define el estado y las salidas tipadas

# app/schemas.py
from typing import Literal

from pydantic import BaseModel, Field, model_validator


class WorkItem(BaseModel):
    id: str
    objective: str
    read_only: bool
    write_globs: list[str] = Field(default_factory=list)
    verification_commands: list[str] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_permissions(self):
        if self.read_only and self.write_globs:
            raise ValueError("read-only tasks cannot declare writes")
        if not self.read_only and not self.write_globs:
            raise ValueError("writable tasks require declared scopes")
        return self


class WorkflowPlan(BaseModel):
    substantive: bool
    reason: str
    success_criteria: list[str] = Field(min_length=1)
    tasks: list[WorkItem] = Field(min_length=1, max_length=4)

    @model_validator(mode="after")
    def validate_wave(self):
        expected = range(2, 5) if self.substantive else range(1, 2)
        if len(self.tasks) not in expected:
            raise ValueError("routine plans need 1 task; substantive plans need 2-4")
        ids = [task.id for task in self.tasks]
        if len(ids) != len(set(ids)):
            raise ValueError("task IDs must be unique")
        return self


class WorkerOutput(BaseModel):
    status: Literal["completed", "blocked"]
    summary: str
    unresolved_risks: list[str] = Field(default_factory=list)


class ReviewOutput(BaseModel):
    candidate_passed: bool
    blocking_issues: list[str] = Field(default_factory=list)
    evidence: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_review(self):
        if self.candidate_passed and self.blocking_issues:
            raise ValueError("passing reviews cannot contain blocking issues")
        return self


class SynthesisOutput(BaseModel):
    summary: str
    acceptance_evidence: list[str] = Field(default_factory=list)
    unresolved_risks: list[str] = Field(default_factory=list)

El grafo debe almacenar por separado la salida generada por el modelo y la identidad gestionada por la infraestructura:

class WorkerRecord(BaseModel):
    task_id: str
    round: int
    workspace_id: str
    base_sha: str
    changed_files: list[str]
    artifact_uris: list[str]
    output: WorkerOutput


class Candidate(BaseModel):
    id: str
    thread_id: str
    round: int
    base_sha: str
    parent_candidate_id: str | None
    candidate_sha: str
    task_ids: list[str]
    workspace_ids: list[str]
    changed_files: list[str]
    diff_uri: str


class TestReport(BaseModel):
    candidate_id: str
    candidate_sha: str
    passed: bool
    exit_codes: dict[str, int]
    artifact_uri: str
    artifact_sha256: str

    @model_validator(mode="after")
    def validate_exit_codes(self):
        expected = bool(self.exit_codes) and all(
            code == 0 for code in self.exit_codes.values()
        )
        if self.passed != expected:
            raise ValueError("passed must match deterministic exit codes")
        return self

Paso 2: Mantén las llamadas al modelo pequeñas y tipadas

# app/model.py
import json
import os
from typing import TypeVar, cast

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

T = TypeVar("T", bound=BaseModel)


def base_model() -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ["AGENT_MODEL_ID"],
        max_retries=2,
    )


async def typed_call(
    schema: type[T],
    *,
    role: str,
    payload: dict,
) -> T:
    model = base_model().with_structured_output(
        schema,
        method="function_calling",
    )
    result = await model.ainvoke(
        [
            SystemMessage(content=role),
            HumanMessage(content=json.dumps(payload)),
        ]
    )
    return cast(T, result)

Usa un modelo que admita llamadas a herramientas y salida estructurada. La ruta del worker requiere específicamente el uso simultáneo de herramientas y una respuesta final estructurada; de lo contrario, selecciona una ToolStrategy explícita de LangChain o implementa el worker como un subgrafo de herramientas dedicado de LangGraph. Mantén contratos de sistema distintos para el planificador, el worker, el reductor, el verificador y el refutador, mientras el código del runtime sigue siendo determinista.

Paso 3: Inyecta un servicio de espacios de trabajo

# app/workspace.py
from typing import Protocol

from app.schemas import Candidate, TestReport, WorkItem, WorkerOutput, WorkerRecord


class WorkspaceService(Protocol):
    async def planner_evidence(self, repository: str, base_sha: str) -> dict: ...
    async def create_worker(
        self,
        *,
        repository: str,
        thread_id: str,
        round: int,
        task: WorkItem,
        expected_base_sha: str,
        parent_candidate_id: str | None,
        idempotency_key: str,
    ) -> tuple[str, list]: ...
    async def inspect_and_freeze(
        self,
        workspace_id: str,
        task: WorkItem,
        output: WorkerOutput,
        *,
        round: int,
    ) -> WorkerRecord: ...
    async def terminate_and_reset_for_retry(self, workspace_id: str) -> None: ...
    async def quarantine(self, workspace_id: str, reason: str) -> None: ...
    async def compose(
        self,
        *,
        thread_id: str,
        round: int,
        records: list[WorkerRecord],
        parent_candidate_id: str | None,
    ) -> Candidate: ...
    async def test(self, candidate: Candidate, commands: list[str]) -> TestReport: ...
    async def evidence(self, candidate: Candidate, report: TestReport) -> dict: ...
    async def promote(
        self,
        candidate: Candidate,
        *,
        approval_record_id: str,
        idempotency_key: str,
    ) -> dict: ...
    async def cleanup(self, thread_id: str) -> None: ...

Este servicio debe obtener la identidad de Git y del sandbox, no de la salida del modelo. Valida los archivos que realmente cambiaron, compone patches, ejecuta comandos, calcula hashes de los artefactos y promueve solo el candidato aprobado.

El servicio de sandbox es responsable de imponer el timeout estricto del worker. Debe revocar el lease, terminar el grupo de procesos o contenedor y recoger los procesos terminados, restaurar el espacio de trabajo a un estado determinista y solo entonces reportar un error de timeout. Esta estructura base evita deliberadamente los timeouts de nodo de LangGraph para workers con efectos secundarios, porque LangGraph 1.2.9 no espera a que termine la limpieza del nodo cancelado antes de propagar NodeTimeoutError.

Paso 4: Construye el grafo

# app/graph.py
import asyncio
import hashlib
import json
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, RetryPolicy, Send, TimeoutPolicy, interrupt
from langchain.agents import create_agent
from pydantic import ValidationError

from app.model import base_model, typed_call
from app.schemas import (
    Candidate,
    ReviewOutput,
    SynthesisOutput,
    TestReport,
    WorkItem,
    WorkerOutput,
    WorkerRecord,
    WorkflowPlan,
)
from app.workspace import WorkspaceService

MAX_REPAIR_ROUNDS = 2


class State(TypedDict, total=False):
    request: str
    repository: str
    base_sha: str
    thread_id: str
    round: int
    original_success_criteria: list[str]
    baseline_commands: list[str]
    plan: dict
    parent_candidate_id: str | None
    worker_records: Annotated[list[dict], operator.add]
    candidate: dict
    test_report: dict
    synthesis: dict
    reviews: Annotated[list[dict], operator.add]
    approval: dict
    wave_failure: dict | None
    final: dict


class WorkerState(TypedDict):
    request: str
    repository: str
    thread_id: str
    round: int
    task: dict
    expected_base_sha: str
    parent_candidate_id: str | None


def retry_transient(exc: Exception) -> bool:
    return isinstance(exc, ConnectionError)


def build_graph(checkpointer, workspaces: WorkspaceService, approvals):
    async def plan_node(state: State) -> dict:
        repo = await workspaces.planner_evidence(
            state["repository"],
            state["base_sha"],
        )
        plan = await typed_call(
            WorkflowPlan,
            role=(
                "Plan one routine task or 2-4 independent tasks. "
                "Declare write scopes and deterministic verification commands."
            ),
            payload={"request": state["request"], "repository": repo},
        )
        return {
            "plan": plan.model_dump(),
            "round": 0,
            "original_success_criteria": plan.success_criteria,
            "baseline_commands": sorted(
                {
                    command
                    for task in plan.tasks
                    for command in task.verification_commands
                }
            ),
            "parent_candidate_id": None,
        }

    def dispatch(state: State):
        plan = WorkflowPlan.model_validate(state["plan"])
        expected_base_sha = (
            state["base_sha"]
            if state["round"] == 0
            else Candidate.model_validate(state["candidate"]).candidate_sha
        )
        return [
            Send(
                "worker",
                {
                    "request": state["request"],
                    "repository": state["repository"],
                    "thread_id": state["thread_id"],
                    "round": state["round"],
                    "task": task.model_dump(),
                    "expected_base_sha": expected_base_sha,
                    "parent_candidate_id": state["parent_candidate_id"],
                },
            )
            for task in plan.tasks
        ]

    async def worker_node(state: WorkerState) -> dict:
        task = WorkItem.model_validate(state["task"])
        workspace_id = None
        try:
            workspace_id, tools = await workspaces.create_worker(
                repository=state["repository"],
                thread_id=state["thread_id"],
                round=state["round"],
                task=task,
                expected_base_sha=state["expected_base_sha"],
                parent_candidate_id=state["parent_candidate_id"],
                idempotency_key=(
                    f'{state["thread_id"]}:{state["round"]}:{task.id}:workspace'
                ),
            )
            coding_agent = create_agent(
                model=base_model(),
                tools=tools,
                response_format=WorkerOutput,
                checkpointer=False,
                system_prompt=(
                    "Complete only the assigned coding task. "
                    "Use only the provided scoped workspace tools."
                ),
            )
            agent_result = await coding_agent.ainvoke(
                {
                    "messages": [
                        {
                            "role": "user",
                            "content": json.dumps(
                                {
                                    "request": state["request"],
                                    "task": task.model_dump(),
                                    "workspace_id": workspace_id,
                                }
                            ),
                        }
                    ]
                }
            )
            output = WorkerOutput.model_validate(
                agent_result["structured_response"]
            )
            record = await workspaces.inspect_and_freeze(
                workspace_id,
                task,
                output,
                round=state["round"],
            )
            return {"worker_records": [record.model_dump()]}
        except asyncio.CancelledError:
            if workspace_id:
                await asyncio.shield(
                    workspaces.terminate_and_reset_for_retry(workspace_id)
                )
            raise
        except Exception as exc:
            if workspace_id:
                if retry_transient(exc):
                    await workspaces.terminate_and_reset_for_retry(workspace_id)
                else:
                    await workspaces.quarantine(workspace_id, str(exc))
            raise

    async def integrate_node(state: State) -> dict:
        plan = WorkflowPlan.model_validate(state["plan"])
        records = [
            WorkerRecord.model_validate(item)
            for item in state["worker_records"]
            if item["round"] == state["round"]
        ]
        expected = {task.id for task in plan.tasks}
        actual = [record.task_id for record in records]
        if (
            len(actual) != len(expected)
            or set(actual) != expected
            or any(record.output.status != "completed" for record in records)
        ):
            return {
                "wave_failure": {
                    "kind": "incomplete_worker_wave",
                    "expected_task_ids": sorted(expected),
                    "actual_task_ids": actual,
                }
            }

        candidate = await workspaces.compose(
            thread_id=state["thread_id"],
            round=state["round"],
            records=records,
            parent_candidate_id=state["parent_candidate_id"],
        )
        expected_base_sha = (
            state["base_sha"]
            if state["round"] == 0
            else Candidate.model_validate(state["candidate"]).candidate_sha
        )
        if (
            candidate.thread_id != state["thread_id"]
            or candidate.round != state["round"]
            or set(candidate.task_ids) != expected
            or set(candidate.workspace_ids)
            != {record.workspace_id for record in records}
            or candidate.parent_candidate_id != state["parent_candidate_id"]
            or candidate.base_sha != expected_base_sha
            or any(record.base_sha != expected_base_sha for record in records)
            or len(candidate.task_ids) != len(set(candidate.task_ids))
            or len(candidate.workspace_ids) != len(set(candidate.workspace_ids))
        ):
            raise RuntimeError("candidate lineage does not match the plan")

        commands = sorted(
            set(state["baseline_commands"])
            | {
                command
                for task in plan.tasks
                for command in task.verification_commands
            }
        )
        report = await workspaces.test(candidate, commands)
        if (
            report.candidate_id != candidate.id
            or report.candidate_sha != candidate.candidate_sha
        ):
            raise RuntimeError("test report is not bound to candidate")
        return {
            "candidate": candidate.model_dump(),
            "test_report": report.model_dump(),
            "wave_failure": None,
        }

    def route_after_integrate(state: State) -> str:
        return "failed" if state.get("wave_failure") else "synthesize"

    async def synthesize_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        evidence = await workspaces.evidence(candidate, report)
        synthesis = await typed_call(
            SynthesisOutput,
            role=(
                "Summarize whether the candidate satisfies the criteria. "
                "Do not hide failed tests or unresolved risks."
            ),
            payload={
                "criteria": state["original_success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence,
            },
        )
        return {"synthesis": synthesis.model_dump()}

    async def review_node(state: State, role: str) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        evidence = await workspaces.evidence(candidate, report)
        review = await typed_call(
            ReviewOutput,
            role=(
                "Verify every criterion using immutable evidence."
                if role == "verifier"
                else "Try to falsify the exact candidate with a blocking example."
            ),
            payload={
                "candidate_sha": candidate.candidate_sha,
                "criteria": state["original_success_criteria"],
                "test_report": report.model_dump(),
                "evidence": evidence,
            },
        )
        return {
            "reviews": [
                {
                    "role": role,
                    "round": state["round"],
                    "candidate_sha": candidate.candidate_sha,
                    **review.model_dump(),
                }
            ]
        }

    async def verifier(state: State) -> dict:
        return await review_node(state, "verifier")

    async def refuter(state: State) -> dict:
        return await review_node(state, "refuter")

    def barrier(_: State) -> dict:
        return {}

    def route_review(state: State) -> str:
        candidate = Candidate.model_validate(state["candidate"])
        report = TestReport.model_validate(state["test_report"])
        current = {
            item["role"]: item
            for item in state["reviews"]
            if (
                item["round"] == state["round"]
                and item["candidate_sha"] == candidate.candidate_sha
            )
        }
        passed = (
            report.passed
            and set(current) == {"verifier", "refuter"}
            and all(item["candidate_passed"] for item in current.values())
            and all(not item["blocking_issues"] for item in current.values())
        )
        if passed:
            return "approval" if candidate.changed_files else "report"
        if state["round"] < MAX_REPAIR_ROUNDS:
            return "repair"
        return "failed"

    async def repair_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        repo = await workspaces.planner_evidence(
            state["repository"],
            candidate.candidate_sha,
        )
        repair = await typed_call(
            WorkflowPlan,
            role=(
                "Create one non-empty bounded repair wave. "
                "Do not change the supplied success criteria."
            ),
            payload={
                "request": state["request"],
                "success_criteria": state["original_success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": state["test_report"],
                "reviews": [
                    item
                    for item in state["reviews"]
                    if item["round"] == state["round"]
                ],
                "repository": repo,
            },
        )
        repair = repair.model_copy(
            update={"success_criteria": state["original_success_criteria"]}
        )
        return {
            "plan": repair.model_dump(),
            "round": state["round"] + 1,
            "parent_candidate_id": candidate.id,
        }

    def approval_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        binding_digest = hashlib.sha256(
            f'{state["thread_id"]}:{candidate.id}:{candidate.candidate_sha}'.encode()
        ).hexdigest()
        approval = interrupt(
            {
                "kind": "merge",
                "thread_id": state["thread_id"],
                "candidate_id": candidate.id,
                "candidate_sha": candidate.candidate_sha,
                "round": state["round"],
                "binding_digest": binding_digest,
                "test_report": state["test_report"],
            }
        )
        return {"approval": approval}

    def route_approval(state: State) -> str:
        return "promote" if state["approval"]["approved"] else "failed"

    async def promote_node(state: State) -> dict:
        candidate = Candidate.model_validate(state["candidate"])
        approval_id = await approvals.verify(
            state["approval"],
            expected_thread_id=state["thread_id"],
            expected_candidate_id=candidate.id,
            expected_candidate_sha=candidate.candidate_sha,
            expected_round=state["round"],
            expected_action="merge",
            expected_binding_digest=hashlib.sha256(
                f'{state["thread_id"]}:{candidate.id}:{candidate.candidate_sha}'.encode()
            ).hexdigest(),
        )
        result = await workspaces.promote(
            candidate,
            approval_record_id=approval_id,
            idempotency_key=f'{state["thread_id"]}:promote:{candidate.candidate_sha}',
        )
        return {"final": {"status": "completed", "result": result}}

    def report_node(state: State) -> dict:
        return {
            "final": {
                "status": "completed",
                "kind": "report",
                "candidate_sha": state["candidate"]["candidate_sha"],
                "synthesis": state["synthesis"],
                "test_report": state["test_report"],
            }
        }

    def failed_node(state: State) -> dict:
        return {
            "final": {
                "status": "failed",
                "round": state["round"],
                "candidate": state.get("candidate"),
                "wave_failure": state.get("wave_failure"),
            }
        }

    async def cleanup_node(state: State) -> dict:
        await workspaces.cleanup(state["thread_id"])
        return {}

    graph = StateGraph(State)
    graph.add_node(
        "plan",
        plan_node,
        retry_policy=RetryPolicy(
            max_attempts=2,
            retry_on=lambda exc: isinstance(exc, ValidationError),
        ),
        timeout=TimeoutPolicy(run_timeout=180),
    )
    graph.add_node(
        "worker",
        worker_node,
        retry_policy=RetryPolicy(
            max_attempts=2,
            retry_on=retry_transient,
        ),
    )
    graph.add_node("integrate", integrate_node)
    graph.add_node("synthesize", synthesize_node)
    graph.add_node("verify", verifier)
    graph.add_node("refute", refuter)
    graph.add_node("barrier", barrier)
    graph.add_node("repair", repair_node)
    graph.add_node("approval", approval_node)
    graph.add_node("promote", promote_node)
    graph.add_node("report", report_node)
    graph.add_node("failed", failed_node)
    graph.add_node("cleanup", cleanup_node)

    graph.add_edge(START, "plan")
    graph.add_conditional_edges("plan", dispatch)
    graph.add_conditional_edges("repair", dispatch)
    graph.add_edge("worker", "integrate")
    graph.add_conditional_edges("integrate", route_after_integrate)
    graph.add_edge("synthesize", "verify")
    graph.add_edge("synthesize", "refute")
    graph.add_edge(["verify", "refute"], "barrier")
    graph.add_conditional_edges("barrier", route_review)
    graph.add_conditional_edges("approval", route_approval)
    graph.add_edge("promote", "cleanup")
    graph.add_edge("report", "cleanup")
    graph.add_edge("failed", "cleanup")
    graph.add_edge("cleanup", END)

    return graph.compile(checkpointer=checkpointer)

El grafo es el único planificador durable de nivel superior y la fuente de verdad. El helper de worker de LangChain ejecuta un ciclo de herramientas efímero, pero no puede elegir espacios de trabajo arbitrarios ni saltarse la cadena de candidato, pruebas y aprobación.

Paso 5: Agrega persistencia durable

Usa InMemorySaver solo para pruebas. Mantén el saver de PostgreSQL abierto durante toda la vida del proceso de la API:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

from app.bootstrap import approvals, workspaces
from app.graph import build_graph


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncPostgresSaver.from_conn_string(
        os.environ["LANGGRAPH_DATABASE_URL"]
    ) as checkpointer:
        app.state.graph = build_graph(checkpointer, workspaces, approvals)
        yield


app = FastAPI(lifespan=lifespan)

El fragmento es adecuado para tráfico bajo. AsyncPostgresSaver 3.1.0 usa un lock a nivel de instancia para las operaciones del cursor, por lo que añadir un pool de conexiones no elimina toda la serialización a nivel del saver. Mide el rendimiento de los checkpoints y escala deliberadamente los workers del grafo y las instancias del saver, o usa la topología de runtime administrada de Agent Server.

Ejecuta await checkpointer.setup() una vez en un job de migración. Habilita LANGGRAPH_STRICT_MSGPACK=true o configura una lista explícita de módulos permitidos para la deserialización de checkpoints.

Cada job de programación independiente necesita un thread_id nuevo que pertenezca al tenant; todas las invocaciones y reanudaciones por aprobación de ese job lo reutilizan:

result = await app.state.graph.ainvoke(
    {
        "request": "Audit all routes for missing authorization.",
        "repository": "github.com/acme/api",
        "base_sha": "abc123",
        "thread_id": thread_id,
        "round": 0,
        "worker_records": [],
        "reviews": [],
    },
    config={"configurable": {"thread_id": thread_id}},
)

Paso 6: Reanuda después de la aprobación

Carga desde el grafo la interrupción pendiente para el tenant autenticado. Persiste un registro de aprobación vinculado a su candidato y a su digest de vinculación; luego reanuda:

from langgraph.types import Command


approval = await approvals.create_from_interrupt(
    pending_interrupt,
    approved=True,
    approved_by=authenticated_user.id,
)

result = await graph.ainvoke(
    Command(resume=approval.model_dump()),
    config={"configurable": {"thread_id": thread_id}},
)

No confíes en un SHA de candidato proporcionado directamente por el cliente. Cárgalo desde el estado del thread pendiente y verifica a quién pertenece.

Opciones de despliegue

FastAPI personalizado

Usa FastAPI cuando quieras tener control total sobre:

• creación de ejecuciones
• autorización de tenants
• streaming
• gestión de colas
• cancelación
• endpoints de aprobación
• retención

Este es el diseño OSS más pequeño, pero requiere más trabajo en la aplicación.

LangSmith Agent Server

Agent Server agrega una cola de ejecución durable, persistencia, streaming, herramientas de despliegue y la opción de separar los workers de API y de cola. La ejecución en un solo host sigue siendo la topología predeterminada para tráfico bajo.

Para un despliegue independiente autohospedado, la documentación actual requiere:

• PostgreSQL 14+
• Redis 5+ o Valkey 8+
• DATABASE_URI y REDIS_URI
• LANGSMITH_API_KEY
• LANGGRAPH_CLOUD_LICENSE_KEY
• salida de red para la verificación de la licencia, salvo que se use la opción documentada sin conexión
• langgraph.json que declare las dependencias y un grafo o factory exportado
• al menos un worker de cola en ejecución continua

Los servidores independientes deben permanecer activos; la documentación prohíbe el escalado serverless a cero porque puede provocar la pérdida de tareas.

Cuando Agent Server administre el despliegue, deja que inyecte el checkpointer y el store de producción en lugar de compilar el grafo desplegado con tu propio saver.

Requisitos de infraestructura

Client / IDE / CI
       |
       v
API + tenant authorization
       |
       v
run queue / graph workers
       |
       +--> PostgreSQL checkpoints
       +--> cancellation and lease signals
       +--> immutable artifact storage
       +--> approval records
       |
       v
sandbox worker pool
  |        |        |
worker   worker   worker
  |        |        |
isolated repository workspaces
       |
       v
Git provider + CI

Redis es opcional en un runtime personalizado pequeño. Resulta útil para la señalización entre réplicas, las colas, la cancelación y los leases. Es obligatorio en la arquitectura documentada de Agent Server independiente.

Reglas de ejecución durable

LangGraph crea checkpoints en los límites de los supersteps. Los nodos pueden volver a ejecutarse después de una recuperación.

Usa claves de idempotencia para:

workspace creation
artifact writes
candidate composition
test runs
approval records
promotion
notifications

Usa:

{thread_id}:{round}:{task_id}:{operation}

Mantén los efectos secundarios no idempotentes después de las interrupciones para aprobación o encapsúlalos en infraestructura de tareas e idempotencia.

Semántica de fallos

El nodo de limpieza final del grafo se encarga de la finalización normal y de los fallos modelados. El supervisor de API/cola debe encargarse de las excepciones no controladas y de la limpieza de recursos huérfanos.

Evaluación

Compara:

1. Un ciclo de modelo y herramientas
2. Un LangGraph secuencial simple
3. Este LangGraph dinámico

Usa:

• tareas rutinarias que deberían crear un worker
• auditorías paralelas de solo lectura
• cambios de código en áreas que no se superponen
• conflictos por superposición
• fallos de pruebas y reparaciones
• rechazo de la aprobación
• reinicio y reanudación
• cancelación durante la ejecución de comandos de shell

Mide la tasa de tareas aceptadas, la precisión de activación, las violaciones de políticas, el número de reparaciones, el costo por tarea aceptada, la tasa de detección de problemas por los revisores, el tiempo de revisión humana y los efectos secundarios duplicados después de reanudar.

Evalúa los recorridos, no solo las respuestas finales. Una respuesta correcta obtenida mediante una escritura no autorizada o una promoción duplicada sigue siendo una ejecución fallida.

Lo que esta estructura base deja a tu cargo

Todavía necesitas:

• una implementación concreta de worktree/sandbox
• herramientas de programación con alcance limitado
• almacenamiento de aprobaciones autenticadas
• rutas de API y autorización de threads
• gestión de colas y backpressure
• retención de artefactos
• cancelación de grupos de procesos
• manifiestos de despliegue
• pruebas de escenarios

El grafo muestra dónde se conectan esos contratos. No convierte un checkout local en un sandbox de seguridad.

Lo que no copia

No copia de Claude Code:

• la heurística privada de activación
• los scripts de JavaScript generados
• el ensamblaje del contexto
• los controles de esfuerzo del proveedor
• la UI del flujo de trabajo
• los mecanismos internos de convergencia

Aproxima el ciclo portable con código de aplicación explícito y comprobable.

Recomendación final

Comienza con:

• un grafo
• un checkpointer persistente
• un servicio de espacios de trabajo
• cuatro workers como máximo
• dos rondas de reparación
• pruebas deterministas
• dos revisiones independientes
• aprobación vinculada al SHA del candidato

La regla de operación es:

LangGraph controla el estado y la planificación. Los modelos proponen. La infraestructura de sandbox ejecuta. El código determinista decide qué se puede promover.

Lecturas relacionadas

• Guía de arquitectura de Claude Code Ultracode para Cursor y agentes personalizados
• Construye un flujo de trabajo de programación tipo Ultracode con Agno y AgentOS
• Agentes de 12 factores en la práctica: LangChain/LangGraph y Agno/AgentOS
• Cómo transmitir LangChain y LangGraph al AI SDK
• Generación segura de código con agentes

Fuentes

• Documentación de LangChain, “Agentes”, consultado el July 15, 2026.
• Documentación de LangChain, “Salida estructurada”, consultado el July 15, 2026.
• Documentación de LangGraph, “Descripción general de la Graph API”, consultado el July 15, 2026.
• Documentación de LangGraph, “Uso de la Graph API”, consultado el July 15, 2026.
• Documentación de LangGraph, “Flujos de trabajo y agentes”, consultado el July 15, 2026.
• Documentación de LangGraph, “Persistencia”, consultado el July 15, 2026.
• Documentación de LangGraph, “Tolerancia a fallos”, consultado el July 15, 2026.
• Documentación de LangGraph, “Interrupciones”, consultado el July 15, 2026.
• Documentación de LangGraph, “Subgrafos”, consultado el July 15, 2026.
• Documentación de LangGraph, “Streaming de eventos”, consultado el July 15, 2026.
• LangGraph, “README del checkpointer de PostgreSQL”, consultado el July 15, 2026.
• Documentación de LangSmith, “Agent Server”, consultado el July 15, 2026.
• Documentación de LangSmith, “Autohospedaje de servidores independientes”, consultado el July 15, 2026.
• Documentación de LangSmith, “Versiones de dependencias para autohospedaje”, consultado el July 15, 2026.
• PyPI, langgraph 1.2.9, July 10, 2026.
• PyPI, langchain 1.3.13, consultado el July 15, 2026.
• PyPI, langchain-openai 1.3.5, consultado el July 15, 2026.
• PyPI, langgraph-checkpoint-postgres 3.1.0, consultado el July 15, 2026.