Límites estructurales para agentes de IA: Dentro del pipeline ResolveFlow
Explica cómo un agente basado en LangGraph garantiza la separación entre el razonamiento y la ejecución mediante verificaciones a nivel de código en lugar de instrucciones en el prompt, incluyendo un error de recuperación que surgió durante el proceso.
La mayoría de las demostraciones de IA con agente siguen el mismo patrón básico: un modelo decide una acción y la ejecuta de inmediato. Se adjunta una herramienta a un prompt, el modelo la invoca y la herramienta se ejecuta sin ninguna verificación adicional. Puede parecer convincente en un breve clip de demostración, pero precisamente este es el escenario que preocupa a quienes consideran la posibilidad de permitir que los sistemas autónomos interactúen con algo importante, ya que a menudo la única protección entre un diagnóstico correcto y una modificación dañina en un sistema en funcionamiento es una frase de advertencia dentro del prompt.
El proyecto descrito aquí se desarrolló para evitar esa dependencia de una única instrucción de precaución.
ResolveFlow acepta un enlace a un issue de GitHub, recopila pruebas que lo respalden, clasifica el problema en una categoría y, a continuación, según dicha categoría, realiza una de tres acciones: lleva a cabo una acción fija e innegociable, inicia una investigación impulsada por un LLM basada en las pruebas obtenidas, o envía el problema directamente a un revisor humano. En lugar de seguir un único ciclo en el que un modelo solicita la llamada a una herramienta, está estructurado como una máquina de estados LangGraph, construida alrededor de una idea central:
El razonamiento y la ejecución permanecen separados debido a la forma en que está construido el sistema, y no porque se le exija seguir alguna convención.
Eso significa que no se trata simplemente de instruir al modelo para que consulte a alguien primero. En cambio, una segunda llamada a un LLM independiente examina y critica el diagnóstico del primer modelo antes de que cualquier humano lo vea. Y la única función en todo el código que tiene permiso para escribir de nuevo en GitHub verifica una marca de aprobación explícita desde dentro de su propia lógica; no porque se espere que el grafo dirija las llamadas de esa manera, sino porque la función en sí se negará a ejecutarse sin dicha marca, incluso si un futuro cambio en el código estableciera una ruta directa que omitiera los pasos habituales.
El resto de esta guía explica cómo se ensambla realmente el sistema, utilizando la implementación real, así como un error que surgió durante el desarrollo. Ese error ofrece una lección útil: un modelo que apunta a un documento fuente real no es lo mismo que un modelo que apunta a una fuente que sea realmente pertinente para la pregunta planteada.
La estructura del pipeline
Se ejecutan seis etapas en secuencia, y solo una de ellas tiene permiso para modificar cualquier elemento fuera del propio pipeline:
- fetch_evidence: llamadas reales a la API REST de GitHub, que obtienen el texto del cuerpo del problema, su secuencia de comentarios y cualquier prueba de CI ejecutada.
- normalize_evidence: el JSON no procesado devuelto por esas llamadas se verifica y se convierte en un objeto IssueEvidence tipado.
- classify: una etapa ligera basada en reglas que no implica ninguna llamada a un LLM.
Las secciones siguientes explican en detalle los aspectos más importantes.
Clasificación: deliberadamente no es una llamada a un LLM
Tener un LLM disponible al instante hace que sea tentador pasar por él todas las decisiones, incluso aquellas que no requieren ese tipo de razonamiento. El paso de clasificación determina hasta qué punto se permite que un problema en particular siga el camino costoso y de mayor riesgo: diagnóstico basado en modelos, solicitudes de recuperación de datos y escrituras finales. Debido a esta función de control, debe ser económico, rápido y completamente predecible por sí mismo:
def classify(state: GraphState) -> dict:
if state["evidence"].has_failing_ci:
return {"classification": "deterministic"}
elif state["evidence"].is_information_sparse:
return {"classification": "ai_investigation"}
else:
return {"classification": "human_review"}
Este paso produce tres resultados posibles, cada uno que exige un nivel diferente de confianza en las etapas posteriores:
- determinista: una verificación de CI que ha fallado es una señal clara y mecánica por sí sola. No hay nada que investigar; el problema simplemente se etiqueta y se pasa a un mantenedor.
Cabe señalar que human_review, y no ai_investigation, es la opción de respaldo. Cada vez que el sistema no puede determinar qué está sucediendo, no intenta improvisar una respuesta ingeniosa.
Diagnóstico: recuperación de información, luego un esquema, no texto libre
Una vez que un problema se dirige al branch ai_investigation, el paso generate_diagnosis busca en un índice de Pinecone que contiene aproximadamente 2,750 fragmentos provenientes de problemas reales y cerrados de cuatro repositorios distintos (facebook/react, langchain-ai/langchain, microsoft/terminal y vercel/next.js). Luego envía esos fragmentos recuperados al LLM como contexto de apoyo para la llamada:
def generate_diagnosis(state: GraphState) -> dict:
evidence = state["evidence"]
query = f"{evidence.title}\n\n{evidence.body}"
snippets = retrieve_evidence(query, k=3)
snippet_block = "\n\n".join(
f"[{s['id']}] (relevance: {s['score']:.2f}) {s['text']}" for s in snippets
)
prompt = (
f"Issue: {evidence.title}\n{evidence.body}\n\n"
f"Comments:\n{chr(10).join(evidence.comments) or '(none)'}\n\n"
f"Evidence snippets (cite by id in square brackets):\n{snippet_block}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Diagnosis)
diagnosis = structured_llm.invoke([("system", _SYSTEM_PROMPT), ("human", prompt)])
return {
"diagnosis": diagnosis,
"retrieved_ids": [s["id"] for s in snippets],
"retrieved_scores": {s["id"]: s["score"] for s in snippets},
}
Algunos detalles de este paso de diagnóstico merecen una atención más detallada.
En primer lugar, la estructura de la salida no es algo que se extrae posteriormente; está garantizada desde el principio. El diagnóstico se define como un modelo Pydantic:
class Diagnosis(BaseModel):
root_cause: str
severity: Literal["low", "medium", "high"]
missing_info: list[str] = Field(default_factory=list)
recommended_next_steps: list[str]
citations: list[str] = Field(
default_factory=list,
description="IDs of retrieved evidence/doc snippets that support each claim above",
)
Al llamar a .with_structured_output(Diagnosis), el modelo se ve obligado a seguir esta estructura exacta mientras genera su respuesta. No hay expresiones regulares que intenten extraer la causa raíz de un bloque de texto posteriormente; cuando la llamada tiene éxito, lo que se devuelve ya es un objeto tipado, no texto que haya que interpretar.
En segundo lugar, las citaciones no dependen del tono ni de la confianza del modelo: deben ser identificadores reales. La instrucción del sistema indica claramente que cualquier cita debe coincidir con uno de los IDs de los fragmentos proporcionados; nada inventado, y nada asociado a una afirmación que el fragmento en realidad no respalde. Esa restricción debería ser suficiente por sí sola, pero no lo es; y ese es precisamente el motivo por el que existe la siguiente etapa del proceso.
Evaluación independiente: el modelo explica, el código decide
Se podría decir que esta es la elección arquitectónica más importante de todo el sistema.
El nodo independent_review realiza una segunda llamada completamente separada a ChatOpenAI, con su propio prompt y sin compartir contexto con la llamada que generó el diagnóstico. Su función es criticar dicho diagnóstico.
Pero aquí está la parte clave: la decisión real de aprobar o escalar nunca se deja en manos del modelo. Se basa en tres valores booleanos calculados con Python estándar, y la salida del LLM se reduce a comentarios legibles para humanos de los cuales la lógica de aprobación en realidad no depende.
def independent_review(state: GraphState) -> dict:
evidence = state["evidence"]
diagnosis = state["diagnosis"]
retrieved_ids = set(state.get("retrieved_ids", []))
retrieved_scores = state.get("retrieved_scores", {})
groundedness_ok = bool(diagnosis.citations) and all(
citation_id in retrieved_ids
and retrieved_scores.get(citation_id, 0.0) >= MIN_RELEVANCE_SCORE
for citation_id in diagnosis.citations
)
risk_ok = diagnosis.severity in _ALLOWED_SEVERITIES # {"low", "medium"}
permission_ok = True # comment/label are the only writes available today
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
reasoning = llm.invoke(
[("system", _SYSTEM_PROMPT), ("human", prompt)]
).content # human-readable critique — not what the gate checks
outcome = "approve" if (groundedness_ok and risk_ok and permission_ok) else "escalate_to_human"
return {"review_result": ReviewResult(
outcome=outcome,
groundedness_ok=groundedness_ok,
risk_ok=risk_ok,
permission_ok=permission_ok,
reasoning=reasoning,
)}
Este es el patrón general que debe seguir cualquier versión fiable de “LLM como juez”: el modelo puede explicarse a sí mismo, pero es el código quien toma la decisión.
Si pides a un modelo que evalúe el trabajo de otro modelo y luego confías simplemente en el veredicto que aparece en su respuesta, en realidad has creado un sistema cuya fiabilidad está limitada por precisamente lo que intentas verificar.
En este diseño, la salida del LLM constituye una narración útil para un lector humano, pero el mecanismo decisivo real no puede ser influenciado hacia un resultado negativo mediante palabras persuasivas, ya que no analiza nada de lo escrito por el modelo para tomar una decisión.
Otro detalle importante: la gravedad alta nunca aparece en _ALLOWED_SEVERITIES. Cualquier diagnóstico marcado como de alta gravedad es elevado automáticamente a un humano, sin importar cuán sólidas sean sus referencias. Ser correcto y ser seguro para aprobar automáticamente no son lo mismo.
El error: “grounded” no es lo mismo que “relevant”
Aquí es donde las cosas se volvieron realmente complicadas en la práctica.
groundedness_ok simplemente comprobaba si un ID de cita coincidía con algo en el conjunto recuperado: un ID real, y no uno fabricado. A simple vista parece una verificación razonable. Pero no era del todo suficiente.
Durante las pruebas con un pequeño corpus de 40 problemas, se sometió a la pipeline un caso real de problema en React con cuerpo realmente vacío (facebook/react#36932, “experimental_taintUniqueValue lanza RangeError para valores binarios grandes”). La recuperación devolvió tres fragmentos, todos legítimos y correctamente identificados; sin embargo, ninguno de ellos tenía relación con este error en particular. Trabajando con ese material escaso, el modelo aún produjo un diagnóstico seguro y detallado que resultó completamente falso: afirmó que se trataba de “un problema de compatibilidad con la extensión React DevTools”. Cada cita superó sin problemas la prueba de fundamentación. El diagnóstico en sí seguía siendo inútil.
La solución fue dejar de considerar “fue recuperado” como sinónimo de “es relevante”. Se actualizó tools/retrieval.py de modo que cada fragmento ahora devuelva su puntuación de similitud coseno junto con su texto:
def retrieve_evidence(query: str, k: int = 3) -> list[dict]:
results = vector_store.similarity_search_with_score(query, k)
return [
{"id": doc.metadata["id"], "text": doc.page_content,
"source": doc.metadata["source"], "score": score}
for doc, score in results
]
Luego verifiqué cómo eran en realidad las puntuaciones de similitud frente al índice actual, en lugar de suponer un umbral arbitrario. Una consulta que coincidía estrechamente con contenido real del corpus obtenía puntuaciones entre 0.53 y 0.63. Una consulta sobre algo completamente ausente de los cuatro repositorios procesados obtenía puntuaciones entre 0.21 y 0.22. Basándome en esa diferencia, independent_review.py ahora impone un umbral mínimo estricto: cada fragmento citado debe superar MIN_RELEVANCE_SCORE = 0.35. Ese umbral se establece deliberadamente más cerca del extremo superior de la diferencia que del centro, ya que permitir que un diagnóstico correcto sea escalado por error conlleva un costo mucho menor que dejar pasar uno incorrecto como aprobado.
Junto con la corrección de la lógica de puntuación, fue necesario ampliar el propio corpus de recuperación. Comenzó con 40 artículos provenientes de un único repositorio, lo que generó aproximadamente 130 fragmentos; fragmentos tan pequeños que un problema real aleatorio a menudo no tenía ningún coincidencia temática genuina para recuperar desde un principio. Esto se amplió a unos 600 artículos extraídos de cuatro repositorios reales, lo que produjo cerca de 2,750 fragmentos.
Con el corpus más amplio en uso, el mismo problema de taintUniqueValue ahora descarta dos informes auténticos de casi duplicados y permite identificar una causa raíz precisa: String.fromCharCode.apply supera el límite de cantidad de argumentos del motor de JavaScript al manejar buffers grandes. Cabe destacar que independent_review sigue elevando este caso para su revisión por parte de humanos, ya que su gravedad se clasifica como alta, y los hallazgos de alta gravedad siempre se elevan por diseño, independientemente del grado de certeza o corrección del diagnóstico. La corrección no exime de pasar por la verificación de riesgos.
La lección más importante aquí es que una cita que apunta a un ID de fragmento real y recuperable es una condición necesaria, pero no suficiente. Afirmar “el modelo citó algo” es una declaración diferente de afirmar “el modelo citó algo verdadero y relevante para este error”; un sistema que solo verifica lo primero parece meticuloso cuando en realidad solo está validando ruido sin sentido.
El paso de aprobación es una pausa real, no un estado de carga cosmético
Cada etapa descrita hasta ahora solo propone una acción. Nada se escribe de vuelta en GitHub hasta un punto específico del proceso. Todo lo anterior: clasificación, diagnóstico, revisión — solo propone; nada se escribe en GitHub hasta este paso único.
El nodo await_approval reúne el comentario exacto (y, cuando corresponde, la etiqueta exacta) que se publicaría, utilizando una lógica idéntica independientemente de si el problema fue dirigido a través del ramo determinista o provino de un diagnóstico ai_investigation aprobado. Luego llama a interrupt() de LangGraph:
def await_approval(state: GraphState) -> dict:
proposed_action = _build_proposed_action(state)
approved = interrupt({
"classification": state["classification"],
"proposed_action": proposed_action,
})
return {"proposed_action": proposed_action, "approved": bool(approved)}
Llamar a interrupt() no se limita a mostrar un indicador de “esperando aprobación” mientras el proceso permanece inactivo; en realidad detiene la ejecución del grafo a mitad de su funcionamiento. Para reanudar esa ejecución posteriormente se necesita una solicitud completamente separada para localizar el mismo hilo pausado y continuar desde donde se interrumpió, utilizando una llamada como result = await compiled_graph.ainvoke(Command(resume=True), config) con el mismo thread_id que utilizó la ejecución original. Para que esto funcione, el estado del grafo debe mantenerse intacto a través de dos solicitudes HTTP independientes, lo que descarta la posibilidad de depender de la memoria interna del proceso para su almacenamiento.
Solo una vez que se completa ese proceso de resumen es cuando se ejecuta el nodo execute. Hay dos detalles importantes que destacar aquí. Primero, la verificación de permisos se encuentra dentro del propio código del nodo y no se expresa únicamente como un borde en el grafo; por lo tanto, si alguna refactorización futura introdujera accidentalmente un borde directo hacia execute, esta verificación seguiría detectándolo. Segundo, execute publica state["proposed_action"] tal como fue aprobado; nunca regenera el comentario posteriormente. Lo que una persona aprueba es exactamente lo que se publica.
def execute(state: GraphState) -> dict:
if not state.get("approved"):
raise PermissionError("execute() called without explicit approval")
evidence = state["evidence"]
action = state["proposed_action"]
token = state.get("github_token")
result = {
"comment": post_comment(evidence.repo, evidence.issue_number,
action["comment"], token=token)
}
if action.get("label"):
try:
result["label"] = add_label(evidence.repo, evidence.issue_number,
action["label"], token=token)
except requests.HTTPError as exc:
result["label_error"] = str(exc)
return {"execution_result": result}
Por qué el backend de almacenamiento para ejecuciones pausadas es más importante de lo que parece
La implementación inicial dependía de SqliteSaver, el cual guarda el estado en un archivo local en el disco del propio backend. Esa configuración funciona sin problemas en la máquina del desarrollador.
Falla en producción de una manera que es fácil pasar por alto: en un host de nivel gratuito como Render, el almacenamiento en disco no es persistente entre reinicios. El proceso se detiene después de un período de inactividad y, con la siguiente solicitud recibida, se reinicia dentro de un contenedor completamente nuevo.
Aquí está lo que sucede concretamente. Una ejecución llega a await_approval y se pausa, esperando que alguien tome acción. Antes de que alguien haga clic en aprobar, la instancia gratuita pasa a estado inactivo y se detiene. La siguiente solicitud crea entonces un contenedor nuevo con una base de datos completamente vacía. Cuando se ejecuta Command(resume=...), ya no hay nada que reanudar: el historial del hilo pausado ha desaparecido sin ningún error ni advertencia.
La solución es AsyncPostgresSaver, respaldado por una base de datos Postgres real (Neon, en esta configuración) que existe independientemente del contenedor en el que se ejecute la aplicación:
async with (
AsyncPostgresSaver.from_conn_string(DATABASE_URL, serde=get_serde()) as saver,
AsyncConnectionPool(DATABASE_URL, open=False,
check=AsyncConnectionPool.check_connection) as pool),
):
Incluso si el contenedor es destruido y reconstruido desde cero, cada hilo pausado sobrevive, siempre y cuando DATABASE_URL siga apuntando a la misma base de datos. La mecánica de pausa y reanudación de interrupt() no cambia en absoluto; solo cambia el lugar donde se encuentra realmente ese estado de pausa.
Cabe destacar por separado que las escrituras realizadas después de la aprobación se efectúan bajo la identidad de quien las aprobó, y no bajo alguna credencial de despliegue compartida. La aplicación en vivo permite que cualquier usuario de GitHub inicie sesión, y cada lectura o escritura para esa operación utiliza el token OAuth propio de dicha persona. Una llamada como post_comment(evidence.repo, evidence.issue_number, action["comment"], token=token) emplea el token del aprobador, y no uno que pertenezca a quien casualmente desplegó la aplicación.
Esto es importante no solo por cuestiones de higiene en la autenticación. Convierte la afirmación “la persona que aprobó esto lo publicó” en algo que GitHub mismo puede confirmar al verificar al autor del comentario, en lugar de ser una simple afirmación hecha por la propia interfaz de la aplicación.
También permite que el sistema de permisos existente de GitHub aplique controles útiles sin necesidad de código adicional: cualquiera que solo esté navegando por un repositorio que no le pertenece puede dejar un comentario, pero adjuntar una etiqueta requiere autorización de triaje o acceso de escritura en dicho repositorio. execute.py gestiona esto de manera adecuada: un intento fallido de agregar una etiqueta se considera un éxito parcial en lugar de hacer fracasar toda la solicitud.
Las llamadas al LLM y al modelo de embedding siguen ejecutándose con las claves API del propietario del despliegue, independientemente de quién las active, y esa es precisamente la razón por la cual existe un límite de uso diario por usuario para mantener esos costos bajo control.
Existe una distinción que merece ser aclarada: la evaluación de regresión y la evaluación de capacidades prueban cosas fundamentalmente diferentes, y calificarlas de la misma manera es un error en el que es fácil caer. La prioridad de enrutamiento dentro de classify(), la lógica de control dentro de independent_review() y la verificación de permisos dentro de execute() tienen cada una exactamente un comportamiento correcto, y la única tasa de aprobación aceptable para ese tipo de pruebas es el 100 por ciento, punto final. Una barrera de seguridad que sea evitada aunque solo sea una vez en cualquier número de ejecuciones constituye un fallo crítico; no es un valor que se pueda suavizar mediante promedios.
Ese estándar es completamente diferente de juzgar si generate_diagnosis produce diagnósticos buenos. Ese tipo de evaluación es calificada por un modelo, realmente mejora con el tiempo y nunca llegaría de forma realista al 100 por ciento. Tratar ambos tipos de verificaciones como si pertenecieran a la misma escala es una trampa común: una barrera de seguridad que “en su mayoría” funciona en realidad no está funcionando como tal.
Qué es realmente real en este momento
Es mejor subestimar esto que exagerarlo, así que aquí está la situación tal como es, en lugar de una presentación engañosa:
La recopilación de evidencias se realiza mediante llamadas reales a GitHub REST, validándose en objetos tipados.
La clasificación se basa en reglas y es determinista, sin participación de modelos de lenguaje grandes.
El diagnóstico proviene de una llamada real a OpenAI que genera un resultado estructurado, basado en búsquedas reales en Pinecone a través de aproximadamente 2,750 fragmentos, que se vuelven a procesar semanalmente.
La revisión independiente es una llamada separada a OpenAI, pero el mecanismo real que impone restricciones es el código, no la opinión del propio modelo.
La etapa de aprobación humana consiste en una pausa real mediante interrupt(), que se reanuda con Command(resume=...), y se verifica byte a byte de extremo a extremo.
Tanto la capa frontal como la posterior están desplegadas —en Vercel y Render respectivamente— de forma completamente asíncrona, y sus estados se almacenan en Postgres.
GitHub OAuth permite que cualquier visitante inicie sesión con su propia cuenta; las escrituras se realizan en nombre de esa persona, con un límite diario establecido.
El conjunto de pruebas de evaluación incluye pruebas de regresión para las barreras de seguridad que r
Esos dos últimos aspectos no se están ocultando; forman parte del plan de desarrollo porque son, de verdad, las cosas más valiosas por desarrollar a continuación, y no porque se hayan pasado por alto.
Lecciones útiles para su propio proyecto
El desarrollo de un agente destinado a actuar en sistemas reales pone de manifiesto varios principios que van más allá de esta herramienta específica para problemas de GitHub:
Mantenga el razonamiento y la ejecución en rutas de código separadas, no solo en prompts distintos. Una instrucción como “siempre pregunte antes de actuar” sigue siendo solo un comportamiento, y los comportamientos tienden a fallar justo en el momento en que más se necesitan. Una función que se niegue categóricamente a ejecutarse a menos que esté activada una bandera verificada constituye un límite real, no una sugerencia.
Cuando diga que un paso de revisión es “independiente”, asegúrese de que eso sea literalmente cierto: una llamada distinta, sin rastro compartido, y un veredicto generado por código y no por texto producido por el propio modelo. El modelo puede explicar su razonamiento, pero no debería ser él quien lo evalúe.
No deje que “el modelo señaló algo real” sustituya a “el modelo señaló algo relevante”. Mida realmente la calidad de la recuperación frente a consultas reales antes de establecer un umbral de similitud.
Si la pausa necesaria para la intervención humana es importante en su diseño, pruebe explícitamente qué ocurre cuando el proceso se reinicia antes de que esa persona responda. El estado en memoria y los discos efímeros parecerán funcionar bien en las pruebas locales, pero luego fallarán justo donde el fallo tiene mayores consecuencias.
Finalmente, sea transparente respecto a qué está realmente terminado y qué sigue siendo una solución temporal. Una tabla de estado que reconozca sus limitaciones genera más confianza que un README que insinúe silenciosamente que todo está listo.
Lecturas relacionadas
- Agentic AI Explained: De los modelos de lenguaje a los agentes autónomos — Una explicación estructurada de cómo los LLM se convierten en sistemas agentes a través de herramientas, memoria, planificación, arquitecturas multiagente e integración MCP.