Módulo 9: Human-in-the-Loop
Proyecto Evolutivo: Aprobaciones Humanas (v4)
Descripción del proyecto
En el Módulo 8, construiste la v3 del AI Research Assistant: un agente con memoria persistente, saludos personalizados, detección de preferencias, y soporte multi-usuario. Si cierras la terminal y la abres de nuevo, el agente recuerda quién eres y qué investigaste. Es un producto con identidad.
Pero toma todas las decisiones solo. Si decide buscar en una API premium que cuesta $5.00, lo hace sin preguntar. Si genera un reporte con un dato incorrecto, lo presenta como verdad. Si su plan de investigación es ineficiente — buscar en 6 fuentes cuando 2 bastarían — no tienes forma de decirle "no, enfócate solo en estas dos." El agente es competente pero no supervisado.
La v4 transforma al Research Assistant de un agente autónomo a un agente supervisado. Antes de buscar, muestra su plan: "Planeo buscar en estas 3 fuentes. Costo estimado: $2.50. ¿Procedo?" Antes de generar el reporte final, muestra un borrador para que revises y des feedback. Si un dato está mal, puedes editarlo directamente en el estado del agente. El resultado es un agente que mantiene al humano en el loop para las decisiones que importan, sin molestar con las que no.
El momento que define esta versión: el agente muestra su plan de investigación, tú eliminas una fuente innecesaria y agregas una que no había considerado. El agente ejecuta con tu plan corregido, genera un borrador, tú señalas un error factual, el agente corrige y produce el reporte final. Esa conversación iterativa entre humano y agente — eso es supervisión efectiva.
Objetivo del proyecto
Evolucionar el AI Research Assistant de v3 (inteligente pero autónomo) a v4 (inteligente y supervisado), agregando aprobación de plan, gate de costo, feedback loop, y edición de estado.
Al completar este proyecto:
- 🔧 Implementarás interrupt en la fase de planificación para mostrar el plan antes de ejecutar
- 🔧 Agregarás un gate de costo que auto-aprueba fuentes gratuitas e interrumpe para fuentes de pago
- 🔧 Crearás un feedback loop donde el usuario revisa el borrador y da retroalimentación (máximo 3 rondas)
- 🔧 Habilitarás edición de estado para que el usuario corrija datos factuales antes del reporte final
- 🔧 Construirás un CLI interactivo que maneja todos los flujos HITL de forma natural
Antes y después
v3 (Módulo 8): inteligente pero autónomo
Usuario: "Investiga AI agents"
Agente: [decide plan internamente — 5 fuentes, $3.50]
Agente: [busca en todas las fuentes sin preguntar]
Agente: [genera reporte con un dato incorrecto]
Agente: "Aquí está tu reporte."
Usuario: "... gastó $3.50 y el segundo hallazgo está mal"
v4 (Este módulo): inteligente y supervisado
Usuario: "Investiga AI agents"
Agente: "Plan: buscar en [web, arXiv, news]. Costo: $0.00. ¿Procedo?"
Usuario: "Agrega academic, quita news"
Agente: [busca en web, arXiv, academic — fuentes gratuitas, auto-approved]
Agente: "También puedo buscar en Premium Research ($2.50). ¿Procedo?"
Usuario: "No, las gratuitas bastan"
Agente: "Borrador del reporte listo. ¿Feedback?"
Usuario: "El segundo hallazgo dice 2023 pero es 2025. Corrige."
Agente: [corrige] → "Reporte final con la corrección."
Especificaciones técnicas
| Componente | Versión | Propósito |
|---|---|---|
| Python | 3.11+ | Runtime |
| LangChain | v1.2+ | Framework de LLMs |
| LangGraph | v1.0+ | Functional API + interrupt + Command |
| langchain-openai | latest | Proveedor de modelos |
| pydantic | v2+ | Modelos structured |
Estructura del proyecto
research-assistant/
├── .env
├── requirements.txt
├── agents/
│ └── researcher.py # MODIFICADO — v4 con interrupts + feedback
├── tools/
│ ├── web_search.py # SIN CAMBIOS (de v2)
│ └── calculator.py # SIN CAMBIOS (de v1)
├── state/
│ └── research_state.py # EXTENDIDO — HumanFeedback model
├── config/
│ └── settings.py # EXTENDIDO — config de costos por fuente
├── memory/
│ └── user_store.py # SIN CAMBIOS (de v3)
├── hitl/ # NUEVO
│ └── risk_assessment.py # Framework de decisión de riesgo
├── utils/
│ ├── retry.py # SIN CAMBIOS (de v2)
│ └── logger.py # SIN CAMBIOS (de v2)
└── main.py # MODIFICADO — CLI con flujos HITL
Paso 1: Framework de riesgo (hitl/risk_assessment.py)
El módulo que decide qué acciones requieren aprobación humana y cuáles se auto-aprueban. Aplica el framework de 4 criterios de la cápsula anterior.
"""
hitl/risk_assessment.py
Framework de decisión de riesgo para el AI Research Assistant v4.
"""
from dataclasses import dataclass
from enum import Enum
class RiskDecision(Enum):
AUTO_APPROVE = "auto_approve"
QUICK_APPROVE = "quick_approve"
FULL_REVIEW = "full_review"
@dataclass
class ActionRisk:
action_name: str
cost: float
reversible: bool
external_impact: bool
confidence: float
@property
def risk_score(self) -> float:
if self.cost < 0.01:
cost_score = 0.0
elif self.cost < 1.0:
cost_score = 0.3
elif self.cost < 10.0:
cost_score = 0.6
else:
cost_score = 1.0
rev_score = 0.0 if self.reversible else 0.8
impact_score = 0.7 if self.external_impact else 0.0
base = max(cost_score, rev_score, impact_score)
confidence_penalty = (1.0 - self.confidence) * 0.5
return min(round(base + confidence_penalty, 2), 1.0)
@property
def decision(self) -> RiskDecision:
score = self.risk_score
if score < 0.3:
return RiskDecision.AUTO_APPROVE
elif score <= 0.7:
return RiskDecision.QUICK_APPROVE
else:
return RiskDecision.FULL_REVIEW
SOURCE_COSTS = {
"web": 0.0,
"academic": 0.0,
"news": 0.0,
"premium_research": 2.50,
"patent_db": 1.00,
"financial_data": 5.00,
}
def assess_source(source_type: str, confidence: float = 0.9) -> ActionRisk:
cost = SOURCE_COSTS.get(source_type, 0.0)
return ActionRisk(
action_name=f"search_{source_type}",
cost=cost,
reversible=True,
external_impact=False,
confidence=confidence,
)
def assess_report_send(recipients: int, confidence: float = 0.8) -> ActionRisk:
return ActionRisk(
action_name="send_report",
cost=0.0,
reversible=False,
external_impact=recipients > 1,
confidence=confidence,
)
def partition_sources(sources: list[str], confidence: float = 0.9) -> dict:
"""Separa fuentes en auto-approved y requires-approval."""
auto = []
needs_approval = []
total_cost = 0.0
for src in sources:
risk = assess_source(src, confidence)
if risk.decision == RiskDecision.AUTO_APPROVE:
auto.append({"source": src, "cost": SOURCE_COSTS.get(src, 0.0)})
else:
needs_approval.append({"source": src, "cost": SOURCE_COSTS.get(src, 0.0), "risk": risk.risk_score})
total_cost += SOURCE_COSTS.get(src, 0.0)
return {
"auto_approved": auto,
"needs_approval": needs_approval,
"total_pending_cost": total_cost,
}Paso 2: Extender modelos de datos (state/research_state.py)
Agrega el modelo HumanFeedback para trackear las interacciones humanas durante la ejecución.
"""
state/research_state.py
Modelos de datos para el AI Research Assistant v4.
"""
from pydantic import BaseModel, Field
from datetime import datetime
class Source(BaseModel):
name: str
source_type: str
content: str
class KeyFinding(BaseModel):
title: str
description: str
confidence: float = Field(ge=0.0, le=1.0)
class SourceStatus(BaseModel):
source_type: str
status: str
attempts: int = Field(default=1)
error: str = Field(default="")
duration_ms: float = Field(default=0)
class UserProfile(BaseModel):
user_id: str
total_sessions: int = 0
top_topics: list[tuple[str, int]] = Field(default_factory=list)
last_topic: str = ""
preferred_format: str = "paragraphs"
is_new_user: bool = True
class HumanFeedback(BaseModel):
"""Registro de una interacción humana durante la ejecución (v4)."""
feedback_type: str
content: str
timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())
round_number: int = 1
class ResearchReport(BaseModel):
topic: str
summary: str
key_findings: list[KeyFinding] = Field(min_length=1)
sources: list[Source] = Field(min_length=1)
sub_queries: list[str]
confidence: float = Field(ge=0.0, le=1.0)
generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
source_availability: list[SourceStatus] = Field(default_factory=list)
version: str = Field(default="v4")
user_id: str = Field(default="anonymous")
session_number: int = Field(default=0)
human_feedback_rounds: int = Field(default=0)
plan_was_edited: bool = Field(default=False)
facts_corrected: int = Field(default=0)
class SubQuery(BaseModel):
query: str
rationale: strPaso 3: Configuración de costos (config/settings.py)
"""
config/settings.py
Configuración del AI Research Assistant v4.
"""
from dotenv import load_dotenv
load_dotenv()
MODEL_NAME = "openai:gpt-4.1-mini"
MODEL_TEMPERATURE = 0.2
MAX_SUB_QUERIES = 4
SEARCH_SOURCES = ["web", "academic", "news"]
RETRY_MAX_ATTEMPTS = 3
RETRY_BASE_DELAY = 1.0
RETRY_MAX_DELAY = 10.0
RETRY_JITTER = True
MIN_SOURCES_FOR_REPORT = 1
# v3: Memory
MAX_CONVERSATION_MESSAGES = 20
MEMORY_PERSIST_FILE = "memory_store.json"
DEFAULT_FORMAT = "paragraphs"
DEFAULT_DETAIL_LEVEL = "standard"
DEFAULT_SOURCES_PRIORITY = ["web", "academic", "news"]
# v4: HITL
MAX_FEEDBACK_ROUNDS = 3
PAID_SOURCES = {"premium_research": 2.50, "patent_db": 1.00, "financial_data": 5.00}
FREE_SOURCES = ["web", "academic", "news"]
AUTO_APPROVE_COST_THRESHOLD = 0.01Paso 4: Agente v4 con HITL (agents/researcher.py)
El agente ahora tiene cuatro puntos de interrupción:
- Plan approval — muestra el plan y espera aprobación/edición
- Cost gate — interrumpe para fuentes de pago
- Draft feedback — muestra borrador y espera feedback (hasta 3 rondas)
- Fact correction — permite editar hallazgos antes del reporte final
"""
agents/researcher.py
AI Research Assistant v4 — human-in-the-loop, supervised execution.
"""
import json
import time
import uuid
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore
from langgraph.types import interrupt
import sys
sys.path.insert(0, ".")
from config.settings import (
MODEL_NAME, MODEL_TEMPERATURE, MAX_SUB_QUERIES,
MIN_SOURCES_FOR_REPORT, MAX_FEEDBACK_ROUNDS,
FREE_SOURCES, PAID_SOURCES,
)
from state.research_state import (
ResearchReport, Source, KeyFinding, SubQuery, SourceStatus, HumanFeedback,
)
from tools.web_search import search_with_retry
from tools.calculator import calculate_confidence
from utils.logger import ResearchLogger
from memory.user_store import (
get_user_profile, get_preferences, record_session, detect_preferences_from_input,
)
from hitl.risk_assessment import partition_sources, assess_source, RiskDecision
model = init_chat_model(MODEL_NAME, temperature=MODEL_TEMPERATURE)
agent_logger = ResearchLogger("research_agent_v4")
@task
def generate_greeting(user_id: str, profile: dict) -> str:
if profile["is_new_user"]:
return "¡Hola! Soy tu asistente de investigación. En esta versión, te consultaré antes de decisiones importantes."
parts = ["¡Bienvenido de vuelta!"]
if profile["last_session"]:
topic = profile["last_session"].get("topic", "")
date = profile["last_session"].get("timestamp", "")[:10]
parts.append(f"Última investigación: '{topic}' ({date}).")
if profile["total_sessions"] > 0:
parts.append(f"Llevas {profile['total_sessions']} sesiones.")
if profile["top_topics"] and profile["top_topics"][0][1] >= 2:
top = profile["top_topics"][0]
parts.append(f"Tema favorito: '{top[0]}' ({top[1]} veces).")
return " ".join(parts)
@task
def decompose_query(topic: str) -> list[dict]:
response = model.invoke(
f"Descompone este tema en {MAX_SUB_QUERIES} sub-preguntas investigables.\n\n"
f"Tema: {topic}\n\n"
f'Responde en JSON: [{{"query": "...", "rationale": "..."}}]\nSolo JSON.'
)
try:
return json.loads(response.content)[:MAX_SUB_QUERIES]
except json.JSONDecodeError:
return [{"query": topic, "rationale": "Fallback"}, {"query": f"avances en {topic}", "rationale": "Tendencias"}]
@task
def plan_research(topic: str, sub_queries: list[dict], available_sources: list[str]) -> dict:
"""Genera el plan de investigación y pide aprobación humana."""
partition = partition_sources(available_sources)
plan = {
"topic": topic,
"sub_queries": [sq["query"] for sq in sub_queries],
"free_sources": [s["source"] for s in partition["auto_approved"]],
"paid_sources": [s for s in partition["needs_approval"]],
"total_free_cost": 0.0,
"total_paid_cost": partition["total_pending_cost"],
}
response = interrupt({
"type": "plan_approval",
"message": (
f"Plan de investigación para '{topic}':\n"
f" Sub-queries: {len(sub_queries)}\n"
f" Fuentes gratuitas: {plan['free_sources']}\n"
f" Fuentes de pago: {[s['source'] for s in plan['paid_sources']]} "
f"(${plan['total_paid_cost']:.2f})\n"
f"¿Procedo? Opciones: approve / edit / cancel"
),
"plan": plan,
})
action = response.get("action", "approve")
if action == "cancel":
return {**plan, "status": "cancelled"}
elif action == "edit":
edited_sources = response.get("sources", plan["free_sources"])
plan["free_sources"] = [s for s in edited_sources if s in FREE_SOURCES]
plan["paid_sources"] = [
{"source": s, "cost": PAID_SOURCES.get(s, 0.0)}
for s in edited_sources if s in PAID_SOURCES
]
plan["total_paid_cost"] = sum(s["cost"] for s in plan["paid_sources"])
plan["was_edited"] = True
return {**plan, "status": "approved_edited"}
else:
return {**plan, "status": "approved"}
@task
def approve_paid_sources(paid_sources: list[dict]) -> list[str]:
"""Gate de costo: pide aprobación individual para cada fuente de pago."""
approved = []
for source_info in paid_sources:
src = source_info["source"]
cost = source_info["cost"]
response = interrupt({
"type": "cost_approval",
"message": f"Buscar en '{src}' cuesta ${cost:.2f}. ¿Aprobar?",
"source": src,
"cost": cost,
})
if response.get("approved", False):
approved.append(src)
print(f" ✅ {src} (${cost:.2f}) — aprobado")
else:
print(f" ❌ {src} (${cost:.2f}) — rechazado")
return approved
@task
def search_all_sources(query: str, sources: list[str], logger: ResearchLogger) -> list[dict]:
start = time.time()
futures = [search_with_retry(query, src, logger) for src in sources]
results = [f.result() for f in futures]
return results
@task
def merge_and_deduplicate(all_results: list[dict]) -> list[dict]:
seen, unique = set(), []
for r in all_results:
if r["search_status"] != "ok":
continue
key = f"{r['source_type']}:{r['content'][:100]}"
if key not in seen:
seen.add(key)
unique.append(r)
unique.sort(key=lambda r: r.get("relevance", 0), reverse=True)
return unique
@task
def synthesize_findings(topic: str, results: list[dict]) -> list[dict]:
text = "\n".join(f"Fuente {i} ({r['source_type']}): {r['content']}" for i, r in enumerate(results, 1))
response = model.invoke(
f"Identifica 3-5 hallazgos clave sobre '{topic}'.\n\nFuentes:\n{text}\n\n"
f'JSON: [{{"title": "...", "description": "...", "confidence": 0.8}}]\nSolo JSON.'
)
try:
return json.loads(response.content)[:5]
except json.JSONDecodeError:
return [{"title": "Hallazgo general", "description": f"Investigación sobre {topic} relevante.", "confidence": 0.6}]
@task
def generate_summary(topic: str, findings: list[dict], source_info: str, user_prefs: dict) -> str:
findings_text = "\n".join(f"- {f['title']}: {f['description']}" for f in findings)
fmt = ""
if user_prefs.get("format") == "bullet_points":
fmt = "Usa bullet points. "
if user_prefs.get("detail_level") == "brief":
fmt += "Sé breve (1-2 oraciones). "
elif user_prefs.get("detail_level") == "detailed":
fmt += "Sé detallado (4-5 oraciones). "
response = model.invoke(
f"Resumen ejecutivo sobre '{topic}'.\nHallazgos:\n{findings_text}\n"
f"Fuentes: {source_info}\n{fmt}Solo el resumen."
)
return response.content.strip()
@task
def collect_draft_feedback(topic: str, summary: str, findings: list[dict], round_num: int) -> dict:
"""Muestra el borrador y recoge feedback del usuario."""
findings_display = "\n".join(
f" {i}. {f['title']} [{f['confidence']:.0%}]: {f['description']}"
for i, f in enumerate(findings, 1)
)
response = interrupt({
"type": "draft_feedback",
"message": (
f"--- Borrador del reporte (ronda {round_num}/{MAX_FEEDBACK_ROUNDS}) ---\n\n"
f"Tema: {topic}\n\n"
f"Resumen:\n{summary}\n\n"
f"Hallazgos:\n{findings_display}\n\n"
f"¿Feedback? Opciones: approve / feedback <texto> / edit_facts"
),
"summary": summary,
"findings": findings,
"round": round_num,
})
return response
@task
def apply_feedback_to_summary(topic: str, current_summary: str, feedback: str, findings: list[dict], user_prefs: dict) -> str:
"""Regenera el resumen incorporando el feedback del usuario."""
findings_text = "\n".join(f"- {f['title']}: {f['description']}" for f in findings)
response = model.invoke(
f"Reescribe este resumen sobre '{topic}' incorporando el feedback del usuario.\n\n"
f"Resumen actual:\n{current_summary}\n\n"
f"Hallazgos:\n{findings_text}\n\n"
f"Feedback del usuario: {feedback}\n\n"
f"Genera solo el resumen mejorado."
)
return response.content.strip()
@task
def collect_fact_corrections(findings: list[dict]) -> list[dict]:
"""Permite al usuario corregir hallazgos factuales."""
response = interrupt({
"type": "fact_correction",
"message": (
"Puedes corregir los hallazgos. Envía una lista de correcciones:\n"
' {"corrections": [{"index": 0, "field": "description", "new_value": "..."}]}\n'
"O envía {\"corrections\": []} si todo está correcto."
),
"findings": findings,
})
corrections = response.get("corrections", [])
corrected = [f.copy() for f in findings]
for correction in corrections:
idx = correction.get("index", -1)
field = correction.get("field", "")
new_value = correction.get("new_value", "")
if 0 <= idx < len(corrected) and field in corrected[idx]:
corrected[idx][field] = new_value
return corrected
def create_research_agent(checkpointer, store):
"""Factory que crea el research agent v4."""
@entrypoint(checkpointer=checkpointer, store=store)
def research_agent(topic: str, *, store: BaseStore) -> dict:
config = entrypoint.get_config()
user_id = config["configurable"].get("user_id", "anonymous")
request_id = uuid.uuid4().hex[:8]
agent_logger.set_request_id(request_id)
pipeline_start = time.time()
profile = get_user_profile(store, user_id)
user_prefs = get_preferences(store, user_id)
greeting = generate_greeting(user_id, profile).result()
print(f"\n{'=' * 60}")
print(f" 🔬 AI Research Assistant v4 — Supervisado")
print(f" {greeting}")
print(f" Tema: {topic} | User: {user_id} | Request: {request_id}")
print(f"{'=' * 60}")
detect_preferences_from_input(store, user_id, topic)
# --- FASE 1: Decompose ---
print(f"\n📋 Descomponiendo tema...")
sub_queries_raw = decompose_query(topic).result()
sub_queries = [SubQuery(**sq) for sq in sub_queries_raw]
for i, sq in enumerate(sub_queries, 1):
print(f" {i}. {sq.query}")
# --- FASE 2: Plan approval (INTERRUPT) ---
all_available = list(FREE_SOURCES) + list(PAID_SOURCES.keys())
plan = plan_research(topic, sub_queries_raw, all_available).result()
if plan.get("status") == "cancelled":
print(f"\n ❌ Investigación cancelada por el usuario.")
return {"error": "Investigación cancelada", "version": "v4"}
was_edited = plan.get("was_edited", False)
active_sources = plan["free_sources"]
print(f"\n Plan {'editado' if was_edited else 'aprobado'}: {active_sources}")
# --- FASE 3: Cost gate for paid sources (INTERRUPT per source) ---
if plan["paid_sources"]:
print(f"\n💰 Evaluando fuentes de pago...")
approved_paid = approve_paid_sources(plan["paid_sources"]).result()
active_sources = active_sources + approved_paid
else:
print(f"\n Sin fuentes de pago en el plan.")
# --- FASE 4: Search ---
print(f"\n🔍 Buscando en {len(active_sources)} fuentes: {active_sources}")
all_raw, source_statuses = [], []
search_futures = [search_all_sources(sq.query, active_sources, agent_logger) for sq in sub_queries]
for i, future in enumerate(search_futures):
results = future.result()
all_raw.extend(results)
ok = sum(1 for r in results if r["search_status"] == "ok")
print(f" Sub-query {i + 1}: {ok}/{len(results)} OK")
for r in results:
source_statuses.append(SourceStatus(
source_type=r["source_type"], status=r["search_status"],
attempts=r.get("attempts", 1), error=r.get("error", ""),
duration_ms=r.get("duration_ms", 0),
))
total_ok = sum(1 for r in all_raw if r["search_status"] == "ok")
total_failed = len(all_raw) - total_ok
if total_ok < MIN_SOURCES_FOR_REPORT:
print(f"\n ❌ Fuentes insuficientes ({total_ok}). Abortando.")
return {"error": f"Solo {total_ok} fuentes. Mínimo: {MIN_SOURCES_FOR_REPORT}", "version": "v4"}
# --- FASE 5: Merge + Synthesize ---
unique = merge_and_deduplicate(all_raw).result()
print(f"\n🔀 {len(all_raw)} raw → {len(unique)} únicos")
findings_raw = synthesize_findings(topic, unique).result()
findings = [KeyFinding(**f) for f in findings_raw]
print(f"🧠 {len(findings)} hallazgos identificados")
source_info = f"{total_ok}/{total_ok + total_failed} fuentes OK"
summary = generate_summary(topic, findings_raw, source_info, user_prefs).result()
# --- FASE 6: Draft feedback loop (INTERRUPT — max 3 rounds) ---
feedback_rounds = 0
human_feedbacks = []
for round_num in range(1, MAX_FEEDBACK_ROUNDS + 1):
print(f"\n📝 Mostrando borrador (ronda {round_num})...")
feedback_response = collect_draft_feedback(topic, summary, findings_raw, round_num).result()
action = feedback_response.get("action", "approve")
if action == "approve":
print(f" ✅ Borrador aprobado en ronda {round_num}.")
break
elif action == "feedback":
feedback_text = feedback_response.get("text", "")
human_feedbacks.append(HumanFeedback(
feedback_type="draft_feedback",
content=feedback_text,
round_number=round_num,
))
print(f" 📝 Feedback recibido: '{feedback_text[:60]}...'")
summary = apply_feedback_to_summary(topic, summary, feedback_text, findings_raw, user_prefs).result()
feedback_rounds += 1
print(f" 🔄 Resumen actualizado.")
elif action == "edit_facts":
print(f"\n ✏️ Modo edición de hechos...")
corrected_findings_raw = collect_fact_corrections(findings_raw).result()
corrections_count = sum(
1 for orig, corr in zip(findings_raw, corrected_findings_raw)
if orig != corr
)
if corrections_count > 0:
findings_raw = corrected_findings_raw
findings = [KeyFinding(**f) for f in findings_raw]
summary = generate_summary(topic, findings_raw, source_info, user_prefs).result()
human_feedbacks.append(HumanFeedback(
feedback_type="fact_correction",
content=f"{corrections_count} hechos corregidos",
round_number=round_num,
))
feedback_rounds += 1
print(f" ✅ {corrections_count} hechos corregidos. Resumen regenerado.")
else:
print(f" Sin correcciones. Continuando.")
else:
print(f"\n ⚠️ Máximo de rondas alcanzado ({MAX_FEEDBACK_ROUNDS}). Usando última versión.")
# --- FASE 7: Build final report ---
avg_rel = sum(r["relevance"] for r in unique) / len(unique) if unique else 0.5
base_conf = calculate_confidence(len(unique), avg_rel, len(findings)).result()
avail_factor = total_ok / (total_ok + total_failed) if (total_ok + total_failed) > 0 else 0.5
confidence = round(base_conf * (0.7 + 0.3 * avail_factor), 2)
sources = [Source(name=r["source_name"], source_type=r["source_type"], content=r["content"]) for r in unique]
facts_corrected = sum(1 for fb in human_feedbacks if fb.feedback_type == "fact_correction")
report = ResearchReport(
topic=topic, summary=summary, key_findings=findings, sources=sources,
sub_queries=[sq.query for sq in sub_queries], confidence=confidence,
source_availability=source_statuses, version="v4",
user_id=user_id, session_number=profile["total_sessions"] + 1,
human_feedback_rounds=feedback_rounds,
plan_was_edited=was_edited,
facts_corrected=facts_corrected,
)
record_session(store, user_id, topic, summary[:200])
pipeline_ms = (time.time() - pipeline_start) * 1000
print(f"\n📄 Reporte v4 generado. Sesión #{report.session_number}.")
print(f" Feedback rounds: {feedback_rounds} | Hechos corregidos: {facts_corrected}")
print(f" Plan editado: {'sí' if was_edited else 'no'} | Tiempo: {pipeline_ms:.0f}ms")
print(f"{'=' * 60}")
return report.model_dump()
return research_agentPaso 5: CLI interactivo con flujos HITL (main.py)
El CLI maneja toda la conversación: recibe los interrupts del agente, los presenta al usuario de forma legible, recoge su respuesta, y reanuda la ejecución.
"""
main.py
CLI del AI Research Assistant v4. Human-in-the-loop interactivo.
"""
import sys
import json
import uuid
sys.path.insert(0, ".")
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from memory.user_store import create_store, persist_store, get_user_profile, get_preferences, save_preference
from agents.researcher import create_research_agent
from config.settings import FREE_SOURCES, PAID_SOURCES
def handle_interrupt(state_snapshot) -> Command:
"""Procesa un interrupt y retorna el Command de respuesta."""
task_data = state_snapshot.tasks
if not task_data:
return Command(resume={"action": "approve"})
for task_item in task_data:
if hasattr(task_item, "interrupts") and task_item.interrupts:
interrupt_data = task_item.interrupts[0].value
interrupt_type = interrupt_data.get("type", "unknown")
print(f"\n{'─' * 58}")
print(f" 🔔 INTERRUPCIÓN: {interrupt_type}")
print(f"{'─' * 58}")
print(f" {interrupt_data.get('message', 'Sin mensaje')}")
print(f"{'─' * 58}")
if interrupt_type == "plan_approval":
return handle_plan_approval(interrupt_data)
elif interrupt_type == "cost_approval":
return handle_cost_approval(interrupt_data)
elif interrupt_type == "draft_feedback":
return handle_draft_feedback(interrupt_data)
elif interrupt_type == "fact_correction":
return handle_fact_correction(interrupt_data)
else:
return Command(resume={"action": "approve"})
return Command(resume={"action": "approve"})
def handle_plan_approval(data: dict) -> Command:
"""Maneja la aprobación del plan de investigación."""
plan = data.get("plan", {})
print(f"\n Fuentes gratuitas: {plan.get('free_sources', [])}")
paid = plan.get("paid_sources", [])
if paid:
print(f" Fuentes de pago: {[s['source'] for s in paid]} (${plan.get('total_paid_cost', 0):.2f})")
while True:
try:
choice = input("\n [approve/edit/cancel] → ").strip().lower()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "cancel"})
if choice in ("approve", "a", "sí", "si", "yes", "y"):
return Command(resume={"action": "approve"})
elif choice in ("cancel", "c", "no", "n"):
return Command(resume={"action": "cancel"})
elif choice.startswith("edit") or choice == "e":
all_sources = list(FREE_SOURCES) + list(PAID_SOURCES.keys())
print(f" Fuentes disponibles: {all_sources}")
try:
sources_input = input(" Escribe las fuentes separadas por coma: ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "cancel"})
edited = [s.strip() for s in sources_input.split(",") if s.strip()]
return Command(resume={"action": "edit", "sources": edited})
else:
print(f" Opción no reconocida: '{choice}'")
def handle_cost_approval(data: dict) -> Command:
"""Maneja la aprobación de una fuente de pago."""
source = data.get("source", "?")
cost = data.get("cost", 0)
print(f"\n Fuente: {source} | Costo: ${cost:.2f}")
while True:
try:
choice = input(f" ¿Aprobar ${cost:.2f}? [sí/no] → ").strip().lower()
except (KeyboardInterrupt, EOFError):
return Command(resume={"approved": False})
if choice in ("sí", "si", "yes", "y", "s"):
return Command(resume={"approved": True})
elif choice in ("no", "n"):
return Command(resume={"approved": False})
else:
print(f" Responde 'sí' o 'no'.")
def handle_draft_feedback(data: dict) -> Command:
"""Maneja el feedback sobre el borrador del reporte."""
round_num = data.get("round", 1)
print(f"\n Ronda {round_num} — opciones:")
print(f" approve → Aprobar borrador actual")
print(f" feedback <tx> → Dar feedback para mejorar")
print(f" edit_facts → Corregir datos factuales")
while True:
try:
choice = input(f"\n → ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "approve"})
if choice.lower() in ("approve", "a", "ok"):
return Command(resume={"action": "approve"})
elif choice.lower().startswith("feedback ") or choice.lower().startswith("fb "):
text = choice.split(" ", 1)[1] if " " in choice else ""
if not text:
try:
text = input(" Tu feedback: ").strip()
except (KeyboardInterrupt, EOFError):
return Command(resume={"action": "approve"})
return Command(resume={"action": "feedback", "text": text})
elif choice.lower() in ("edit_facts", "edit", "ef"):
return Command(resume={"action": "edit_facts"})
else:
print(f" Opción no reconocida. Usa: approve, feedback <texto>, edit_facts")
def handle_fact_correction(data: dict) -> Command:
"""Maneja la corrección de hechos."""
findings = data.get("findings", [])
print(f"\n Hallazgos actuales:")
for i, f in enumerate(findings):
print(f" [{i}] {f.get('title', '?')}: {f.get('description', '?')[:70]}...")
print(f"\n Para corregir, escribe: <índice> <nuevo texto>")
print(f" Escribe 'done' cuando termines.")
corrections = []
while True:
try:
line = input(" corrección → ").strip()
except (KeyboardInterrupt, EOFError):
break
if line.lower() in ("done", "d", "listo", ""):
break
parts = line.split(" ", 1)
if len(parts) == 2 and parts[0].isdigit():
idx = int(parts[0])
if 0 <= idx < len(findings):
corrections.append({"index": idx, "field": "description", "new_value": parts[1]})
print(f" ✅ Hallazgo [{idx}] marcado para corrección.")
else:
print(f" ⚠️ Índice {idx} fuera de rango (0-{len(findings)-1})")
else:
print(f" Formato: <índice> <nuevo texto>")
return Command(resume={"corrections": corrections})
def format_report(report: dict) -> str:
if "error" in report:
return f"\n ❌ {report['error']}"
lines = [
"", "╔" + "═" * 58 + "╗",
"║" + f" 📄 REPORTE v4 | {report.get('user_id')} | Sesión #{report.get('session_number')}".center(58) + "║",
"╚" + "═" * 58 + "╝",
f"\n📌 Tema: {report['topic']}",
f"🎯 Confianza: {report['confidence']:.0%}",
]
hitl_info = []
if report.get("plan_was_edited"):
hitl_info.append("Plan editado por usuario")
if report.get("human_feedback_rounds", 0) > 0:
hitl_info.append(f"{report['human_feedback_rounds']} rondas de feedback")
if report.get("facts_corrected", 0) > 0:
hitl_info.append(f"{report['facts_corrected']} hechos corregidos")
if hitl_info:
lines.append(f"👤 Supervisión: {' | '.join(hitl_info)}")
lines.extend([f"\n{'─' * 60}", "📋 RESUMEN", f"{'─' * 60}", report["summary"]])
lines.extend([f"\n{'─' * 60}", "💡 HALLAZGOS", f"{'─' * 60}"])
for i, f in enumerate(report["key_findings"], 1):
lines.append(f" {i}. {f['title']} [{f['confidence']:.0%}]")
lines.append(f" {f['description']}")
lines.extend([f"\n{'─' * 60}", f"📚 FUENTES ({len(report['sources'])})", f"{'─' * 60}"])
for s in report["sources"]:
lines.append(f" • [{s['source_type'].upper()}] {s['name']}")
if report.get("source_availability"):
lines.extend([f"\n{'─' * 60}", "🔌 DISPONIBILIDAD", f"{'─' * 60}"])
for sa in report["source_availability"]:
icon = "✅" if sa["status"] == "ok" else "❌"
retry = f" ({sa['attempts']} intentos)" if sa["attempts"] > 1 else ""
lines.append(f" {icon} {sa['source_type']}: {sa['status']}{retry}")
lines.append(f"\n{'═' * 60}")
return "\n".join(lines)
def run_research_with_hitl(agent, topic: str, user_id: str, store):
"""Ejecuta una investigación completa, manejando todos los interrupts."""
thread_id = f"v4-{user_id}-{uuid.uuid4().hex[:8]}"
config = {"configurable": {"thread_id": thread_id, "user_id": user_id}}
result = agent.invoke(topic, config)
while True:
state = agent.get_state(config)
if not state.next:
break
command = handle_interrupt(state)
result = agent.invoke(command, config)
return result
def run_interactive():
print("=" * 60)
print(" 🔬 AI Research Assistant v4 — Human-in-the-Loop")
print("=" * 60)
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)
current_user = None
print("\n Comandos: user <nombre> | profile | pref <k> <v> | salir\n")
while True:
if not current_user:
try:
current_user = input("👤 Usuario: ").strip()
except (KeyboardInterrupt, EOFError):
break
if not current_user:
continue
profile = get_user_profile(store, current_user)
if profile["is_new_user"]:
print(f" ¡Bienvenido, {current_user}! Primera vez aquí.\n")
else:
print(f" ¡Bienvenido, {current_user}! ({profile['total_sessions']} sesiones)")
if profile["last_session"]:
print(f" Última: '{profile['last_session']['topic']}' ({profile['last_session']['timestamp'][:10]})\n")
continue
try:
user_input = input(f"🔎 [{current_user}] ").strip()
except (KeyboardInterrupt, EOFError):
break
if not user_input:
continue
if user_input.lower() in ("salir", "exit", "quit"):
break
if user_input.lower().startswith("user "):
current_user = user_input[5:].strip()
profile = get_user_profile(store, current_user)
status = "nuevo" if profile["is_new_user"] else f"{profile['total_sessions']} sesiones"
print(f" Cambiado a {current_user} ({status})\n")
continue
if user_input.lower() == "profile":
profile = get_user_profile(store, current_user)
prefs = get_preferences(store, current_user)
print(f"\n Sesiones: {profile['total_sessions']}")
for t, c in profile["top_topics"]:
print(f" - {t} ({c}x)")
print(f" Prefs: {json.dumps(prefs, ensure_ascii=False)}\n")
continue
if user_input.lower().startswith("pref "):
parts = user_input.split(maxsplit=2)
if len(parts) == 3:
save_preference(store, current_user, parts[1], parts[2], "Manual")
print(f" ✓ {parts[1]} = {parts[2]}\n")
continue
try:
report = run_research_with_hitl(agent, user_input, current_user, store)
print(format_report(report))
except Exception as e:
print(f"\n ❌ Error: {e}\n")
print("\n Guardando memoria...")
persist_store(store)
print(" ¡Hasta luego!")
if __name__ == "__main__":
run_interactive()Ejecución: la experiencia supervisada completa
Sesión típica con los 4 tipos de HITL
cd research-assistant
python main.py============================================================
🔬 AI Research Assistant v4 — Human-in-the-Loop
============================================================
[Memory] Cargados 5 items desde memory_store.json
👤 Usuario: mike
¡Bienvenido, mike! (3 sesiones)
Última: 'RAG techniques' (2026-03-07)
🔎 [mike] AI agents frameworks comparison
============================================================
🔬 AI Research Assistant v4 — Supervisado
¡Bienvenido de vuelta! Última investigación: 'RAG techniques' (2026-03-07). Llevas 3 sesiones.
Tema: AI agents frameworks comparison | User: mike | Request: a1b2c3d4
============================================================
📋 Descomponiendo tema...
1. What are the main AI agent frameworks available?
2. How do LangGraph, CrewAI, and AutoGen compare?
3. What are the production considerations for each?
4. What are real-world use cases for each framework?
──────────────────────────────────────────────────────
🔔 INTERRUPCIÓN: plan_approval
──────────────────────────────────────────────────────
Plan de investigación para 'AI agents frameworks comparison':
Sub-queries: 4
Fuentes gratuitas: [web, academic, news]
Fuentes de pago: [premium_research] ($2.50)
¿Procedo? Opciones: approve / edit / cancel
──────────────────────────────────────────────────────
Fuentes gratuitas: ['web', 'academic', 'news']
Fuentes de pago: ['premium_research'] ($2.50)
[approve/edit/cancel] → edit
Fuentes disponibles: ['web', 'academic', 'news', 'premium_research', 'patent_db', 'financial_data']
Escribe las fuentes separadas por coma: web, academic
Plan editado: ['web', 'academic']
Sin fuentes de pago en el plan.
🔍 Buscando en 2 fuentes: ['web', 'academic']
Sub-query 1: 2/2 OK
Sub-query 2: 2/2 OK
Sub-query 3: 1/2 OK
Sub-query 4: 2/2 OK
🔀 16 raw → 7 únicos
🧠 4 hallazgos identificados
📝 Mostrando borrador (ronda 1)...
──────────────────────────────────────────────────────
🔔 INTERRUPCIÓN: draft_feedback
──────────────────────────────────────────────────────
--- Borrador del reporte (ronda 1/3) ---
Tema: AI agents frameworks comparison
Resumen:
Los frameworks principales para agentes de IA incluyen LangGraph,
CrewAI y AutoGen. LangGraph destaca por su control granular...
Hallazgos:
1. LangGraph ofrece control granular [85%]: Permite definir...
2. CrewAI simplifica multi-agent [80%]: Framework orientado a...
3. AutoGen lidera en conversación [75%]: Diseñado por Microsoft...
4. Producción favorece LangGraph [82%]: En entornos de producción...
¿Feedback? Opciones: approve / feedback <texto> / edit_facts
──────────────────────────────────────────────────────
Ronda 1 — opciones:
approve → Aprobar borrador actual
feedback <tx> → Dar feedback para mejorar
edit_facts → Corregir datos factuales
→ feedback Agrega una comparación más directa entre los tres. Menciona que CrewAI es más nuevo y tiene menos comunidad.
📝 Feedback recibido: 'Agrega una comparación más directa entre los tres...'
🔄 Resumen actualizado.
📝 Mostrando borrador (ronda 2)...
──────────────────────────────────────────────────────
🔔 INTERRUPCIÓN: draft_feedback
──────────────────────────────────────────────────────
(borrador mejorado con la comparación directa)
→ edit_facts
✏️ Modo edición de hechos...
──────────────────────────────────────────────────────
🔔 INTERRUPCIÓN: fact_correction
──────────────────────────────────────────────────────
Hallazgos actuales:
[0] LangGraph ofrece control granular: Permite definir flujos...
[1] CrewAI simplifica multi-agent: Framework orientado a roles...
[2] AutoGen lidera en conversación: Diseñado por Microsoft en 2023...
[3] Producción favorece LangGraph: En entornos de producción...
Para corregir, escribe: <índice> <nuevo texto>
Escribe 'done' cuando termines.
corrección → 2 Diseñado por Microsoft Research, con versión 0.4 lanzada en 2025. Soporta patrones de agente conversacional y herramientas extensibles.
✅ Hallazgo [2] marcado para corrección.
corrección → done
✅ 1 hechos corregidos. Resumen regenerado.
📝 Mostrando borrador (ronda 3)...
→ approve
✅ Borrador aprobado en ronda 3.
📄 Reporte v4 generado. Sesión #4.
Feedback rounds: 2 | Hechos corregidos: 1
Plan editado: sí | Tiempo: 45230ms
============================================================
╔══════════════════════════════════════════════════════════╗
║ 📄 REPORTE v4 | mike | Sesión #4 ║
╚══════════════════════════════════════════════════════════╝
📌 Tema: AI agents frameworks comparison
🎯 Confianza: 78%
👤 Supervisión: Plan editado por usuario | 2 rondas de feedback | 1 hechos corregidos
...
La sesión completa incluye los 4 tipos de HITL:
- ✅ Plan approval → el usuario editó las fuentes (quitó
news, quitópremium_research) - ✅ Cost gate → no se activó porque el plan editado no tiene fuentes de pago
- ✅ Draft feedback → el usuario dio feedback textual que mejoró el resumen
- ✅ Fact correction → el usuario corrigió un dato de AutoGen que estaba desactualizado
Criterios de éxito
- ✅ Plan approval funciona — el agente muestra el plan, el usuario puede aprobar, editar fuentes, o cancelar
- ✅ Cost gate funciona — fuentes gratuitas se auto-aprueban, fuentes de pago interrumpen con estimación de costo
- ✅ Feedback loop mejora la calidad — el usuario da feedback textual, el agente regenera el resumen incorporándolo
- ✅ Edición de estado corrige errores factuales — el usuario edita hallazgos directamente, el resumen se regenera con los datos corregidos
- ✅ CLI maneja todos los flujos — la interacción es natural, con prompts claros y opciones entendibles
- ✅ Metadata de supervisión en el reporte — el reporte incluye
human_feedback_rounds,plan_was_edited,facts_corrected - ✅ Memoria persiste entre sesiones — los datos del v3 (preferencias, historial) siguen funcionando
Escenarios de prueba
Test 1: Plan aprobado sin cambios
🔎 [mike] quantum computing trends
→ Plan: [web, academic, news] + [premium_research $2.50]
→ approve
→ Cost gate: premium_research $2.50 → no
→ Borrador → approve
Resultado: reporte con 3 fuentes gratuitas, sin rondas de feedback, plan no editado.
Test 2: Plan editado + fuente de pago aprobada
🔎 [mike] patent analysis for AI chips
→ Plan: [web, academic, news] + [premium_research, patent_db]
→ edit → web, academic, patent_db
→ Cost gate: patent_db $1.00 → sí
→ Borrador → approve
Resultado: reporte con academic + web + patent_db. Plan editado. Costo: $1.00.
Test 3: Investigación cancelada
🔎 [mike] something I changed my mind about
→ Plan: [web, academic, news]
→ cancel
Resultado: {"error": "Investigación cancelada", "version": "v4"}
Test 4: Feedback loop completo (3 rondas)
🔎 [mike] deep learning optimization techniques
→ approve plan
→ Borrador ronda 1 → feedback "Más enfoque en hardware-aware optimization"
→ Borrador ronda 2 → feedback "Agrega mention de quantization"
→ Borrador ronda 3 → approve
Resultado: reporte mejorado por 2 rondas de feedback. 3ra ronda aprobada.
Test 5: Corrección factual
🔎 [mike] LLM benchmarks 2025
→ approve plan
→ Borrador → edit_facts
→ Hallazgo [1] dice "GPT-4 lidera en MMLU" → corregir a "Claude 3.5 Sonnet lidera en MMLU desde Oct 2024"
→ done → approve
Resultado: reporte con 1 hecho corregido reflejado en el resumen final.
Test 6: Múltiples fuentes de pago — aprobación parcial
🔎 [mike] competitive analysis SaaS pricing
→ Plan incluye [premium_research $2.50, financial_data $5.00]
→ approve plan
→ Cost gate: premium_research $2.50 → sí
→ Cost gate: financial_data $5.00 → no
Resultado: reporte usa premium_research pero no financial_data. Costo: $2.50 en vez de $7.50.
Errores comunes
1. interrupt() no pausa — el agente ejecuta sin detenerse
Causa: El grafo no tiene checkpointer configurado. Sin checkpointer, interrupt() no puede guardar el estado y no funciona.
Solución: Verifica que @entrypoint(checkpointer=checkpointer) está configurado y que checkpointer no es None:
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store)2. Command(resume=...) no reanuda — el agente empieza de cero
Causa: El thread_id en la config de resume no coincide con el de la invocación original.
Solución: Usa exactamente la misma config para resume:
config = {"configurable": {"thread_id": "v4-mike-abc123", "user_id": "mike"}}
result = agent.invoke(topic, config)
# ... obtener interrupt ...
result = agent.invoke(Command(resume=response), config) # mismo config3. El feedback loop no termina — ciclo infinito
Causa: La condición de salida del loop no maneja todos los casos, o el action del response no coincide con los valores esperados.
Solución: Siempre incluye un else o un break por defecto, y limita las rondas:
for round_num in range(1, MAX_FEEDBACK_ROUNDS + 1):
response = collect_feedback(...).result()
if response.get("action") == "approve":
break
# ... handle feedback ...
else:
pass # max rondas alcanzado, usar última versión4. Las correcciones de hechos no se reflejan en el resumen
Causa: Después de corregir los hallazgos, no se regenera el resumen con los datos nuevos.
Solución: Siempre regenera el resumen después de editar hallazgos:
corrected = collect_fact_corrections(findings_raw).result()
if corrections_count > 0:
findings_raw = corrected
summary = generate_summary(topic, findings_raw, ...).result() # regenerar5. state.tasks está vacío al intentar leer el interrupt
Causa: Estás leyendo el estado antes de que el agente haya llegado al punto de interrupt, o el interrupt ya fue resuelto.
Solución: Lee el estado solo cuando state.next indica que el agente está pausado:
state = agent.get_state(config)
if state.next: # hay un nodo pendiente → hay interrupt
for task in state.tasks:
if hasattr(task, "interrupts") and task.interrupts:
data = task.interrupts[0].value6. El CLI no distingue entre tipos de interrupt
Causa: Todos los interrupts se manejan igual, sin inspeccionar el type del payload.
Solución: Cada interrupt envía un type en su payload. Usa un dispatcher:
interrupt_type = data.get("type", "unknown")
if interrupt_type == "plan_approval":
return handle_plan_approval(data)
elif interrupt_type == "cost_approval":
return handle_cost_approval(data)7. La memoria del v3 no funciona después de agregar HITL
Causa: El store o el checkpointer no se pasan correctamente al factory create_research_agent().
Solución: Verifica que ambos se crean y se pasan:
store = create_store()
checkpointer = MemorySaver()
agent = create_research_agent(checkpointer, store) # ambos argumentos8. Los costos de fuentes pagadas no coinciden con la realidad
Causa: SOURCE_COSTS en risk_assessment.py no refleja los precios actuales de las APIs.
Solución: Mantén una sola fuente de verdad para los costos:
# config/settings.py — fuente de verdad
PAID_SOURCES = {"premium_research": 2.50, "patent_db": 1.00}
# hitl/risk_assessment.py — importa de config
from config.settings import PAID_SOURCES
SOURCE_COSTS = {**{s: 0.0 for s in FREE_SOURCES}, **PAID_SOURCES}Lo que viene: Módulo 10 — Multi-Agent Systems
Tu Research Assistant v4 es un agente supervisado: muestra planes, pide aprobación para acciones costosas, acepta feedback iterativo, y permite corregir datos factuales. Es la experiencia completa de human-in-the-loop con un solo agente.
Pero es un solo agente que hace todo: planifica, busca, sintetiza, escribe. ¿Qué pasa cuando la tarea es demasiado grande o especializada para uno solo? ¿Qué pasa cuando necesitas un agente que busque, otro que analice, y un tercero que escriba — cada uno con sus propias herramientas y modelo?
El Módulo 10 escala de un agente supervisado a múltiples agentes coordinados. Uno planifica, otro ejecuta búsquedas, otro sintetiza. El humano supervisa al "manager" que coordina a los demás. Las mismas técnicas de HITL que aprendiste — aprobación, feedback, edición de estado — ahora aplican a nivel de equipo: ¿qué agente necesita supervisión? ¿Cuál puede operar autónomo? ¿Cómo coordinas las aprobaciones cuando hay 5 agentes trabajando en paralelo?
Recursos del proyecto
- LangGraph Human-in-the-Loop — Conceptos oficiales de HITL
- interrupt() API Reference — Reference de la función interrupt
- LangGraph Command — Uso de Command para resume
- How to edit graph state — Edición de estado durante ejecución
- LangGraph Functional API + HITL — Interrupt en @entrypoint y @task
- How to review tool calls — Patrón de revisión de herramientas
Módulo 9 — LangChain & LangGraph: From Chains to Agents