Módulo 6: Functional API

Proyecto Evolutivo: Research Agent Base (v1)

Descripción del proyecto

En las siete cápsulas anteriores aprendiste la Functional API de LangGraph: @entrypoint para definir workflows como funciones, @task para tareas checkpointables, control flow nativo con loops y condicionales, la comparación profunda entre Graph API y Functional API, patterns avanzados como futures para ejecución paralela, y cómo combinar ambas APIs cuando el caso lo requiere. Cada concepto lo viste de forma individual. Ahora vas a combinar todo en un sistema real.

Pero este no es un mini-proyecto más. Este es el inicio del AI Research Assistant — un proyecto que vas a construir, iterar, y escalar durante los próximos 6 módulos. Lo que construyas hoy es v1: un agente funcional que recibe un tema de investigación, lo descompone en sub-preguntas, busca información en múltiples fuentes en paralelo, y genera un reporte estructurado. Es simple, pero funciona end-to-end. Y cada pieza está diseñada para evolucionar.

En el Módulo 7, le agregarás retry logic y error handling robusto. En el Módulo 8, memoria persistente para que recuerde investigaciones anteriores. En el Módulo 9, aprobaciones humanas antes de acciones costosas. En el Módulo 10, múltiples agentes especializados coordinados por un supervisor. En el Módulo 11, planning autónomo con Deep Agents. En el Módulo 12, observabilidad completa con LangSmith. Cada módulo agrega una capa sobre lo que construyes hoy.

Construye la v1 bien. Es la base sobre la que todo lo demás se apoya.

Objetivo del proyecto

Construir un agente de investigación funcional con la Functional API de LangGraph que recibe un tema, lo descompone en sub-queries, ejecuta búsquedas en paralelo en múltiples fuentes, y genera un reporte estructurado con Pydantic.

Al completar este proyecto:

🔧 Sabrás diseñar modelos de datos con Pydantic para structured output de agentes
🔧 Implementarás @task functions para descomposición, búsqueda y síntesis
🔧 Crearás un @entrypoint que orquesta el flujo completo de investigación
🔧 Usarás el pattern de Futures para búsqueda paralela en múltiples fuentes
🔧 Generarás streaming de progreso durante la ejecución
🔧 Construirás un CLI simple para interactuar con el agente

Especificaciones técnicas

Stack tecnológico

Componente	Versión	Propósito
Python	3.11+	Runtime
LangChain	v1.2+	Framework de LLMs
LangGraph	v1.0+	Functional API (@entrypoint, @task)
langchain-openai	latest	Proveedor de modelos
pydantic	v2+	Modelos de datos structured
python-dotenv	latest	Variables de entorno

Setup inicial

pip install langchain langgraph langchain-openai pydantic python-dotenv

Crea un archivo .env en la raíz de tu proyecto:

# .env
OPENAI_API_KEY=sk-...

Estructura del proyecto

research-assistant/
├── .env                        # API key
├── requirements.txt            # Dependencias
├── agents/
│   └── researcher.py           # @entrypoint — agente principal
├── tools/
│   ├── web_search.py           # Mock web search (@task)
│   └── calculator.py           # Calculator tool (@task)
├── state/
│   └── research_state.py       # Modelos Pydantic para el reporte
├── config/
│   └── settings.py             # Configuración del proyecto
└── main.py                     # CLI entrypoint

Esta estructura parece excesiva para un proyecto v1. No lo es. Cada directorio tiene un propósito que se revelará en módulos posteriores:

agents/ — En M10 tendrás analyst.py, writer.py, supervisor.py aquí
tools/ — En M7 agregarás document_reader.py con retry logic
state/ — En M8 los modelos se extenderán con campos de memoria
config/ — En M12 tendrás configuración de LangSmith y rate limiting

Paso 1: Modelos de datos (state/research_state.py)

El reporte de investigación necesita estructura. No es un string libre — es un objeto con campos tipados que cualquier sistema downstream puede consumir. Pydantic te da validación, serialización, y documentación automática.

"""
state/research_state.py
Modelos de datos para el AI Research Assistant.
"""

from pydantic import BaseModel, Field
from datetime import datetime


class Source(BaseModel):
    """Una fuente de información encontrada durante la investigación."""
    name: str = Field(description="Nombre de la fuente")
    source_type: str = Field(description="Tipo: web, academic, news")
    content: str = Field(description="Contenido extraído de la fuente")


class KeyFinding(BaseModel):
    """Un hallazgo clave identificado en la investigación."""
    title: str = Field(description="Título del hallazgo")
    description: str = Field(description="Descripción del hallazgo")
    confidence: float = Field(
        description="Confianza en el hallazgo (0.0 a 1.0)",
        ge=0.0,
        le=1.0,
    )


class ResearchReport(BaseModel):
    """Reporte estructurado de investigación."""
    topic: str = Field(description="Tema investigado")
    summary: str = Field(description="Resumen ejecutivo (2-3 oraciones)")
    key_findings: list[KeyFinding] = Field(
        description="Hallazgos principales",
        min_length=1,
    )
    sources: list[Source] = Field(
        description="Fuentes consultadas",
        min_length=1,
    )
    sub_queries: list[str] = Field(
        description="Sub-preguntas generadas para la investigación",
    )
    confidence: float = Field(
        description="Confianza general del reporte (0.0 a 1.0)",
        ge=0.0,
        le=1.0,
    )
    generated_at: str = Field(
        default_factory=lambda: datetime.now().isoformat(),
        description="Timestamp de generación",
    )


class SubQuery(BaseModel):
    """Una sub-pregunta generada a partir del tema principal."""
    query: str = Field(description="La sub-pregunta")
    rationale: str = Field(description="Por qué esta pregunta es relevante")

Cada modelo tiene un propósito claro. ResearchReport es el output final del agente. Source y KeyFinding son los building blocks del reporte. SubQuery estructura la descomposición del tema.

Paso 2: Configuración (config/settings.py)

"""
config/settings.py
Configuración del AI Research Assistant.
"""

from dotenv import load_dotenv
load_dotenv()

MODEL_NAME = "openai:gpt-4.1-mini"
MODEL_TEMPERATURE = 0.2

MAX_SUB_QUERIES = 4
MAX_SOURCES_PER_QUERY = 3

SEARCH_SOURCES = ["web", "academic", "news"]

Centralizar la configuración hace que los cambios sean fáciles. Cuando en M12 agregues configuración de LangSmith y rate limits, todo estará en un solo lugar.

Paso 3: Tools — búsqueda y cálculo (tools/)

Las tools del agente son @task functions. En v1 usamos búsqueda mock — no dependemos de APIs externas que pueden fallar, cambiar, o costar dinero. En M7 agregarás búsqueda real con retry logic.

tools/web_search.py

"""
tools/web_search.py
Mock web search tool para el AI Research Assistant.
En M7 se reemplazará con búsqueda real + retry logic.
"""

import hashlib
from langgraph.func import task


MOCK_RESULTS = {
    "web": {
        "default": (
            "Múltiples fuentes web coinciden en que {query} es un tema de creciente "
            "interés. Expertos destacan avances significativos en los últimos 2 años. "
            "Las aplicaciones prácticas incluyen automatización, análisis de datos "
            "y generación de contenido."
        ),
    },
    "academic": {
        "default": (
            "Investigación reciente (2025-2026) muestra que {query} tiene fundamentos "
            "teóricos sólidos respaldados por múltiples estudios peer-reviewed. "
            "Los resultados experimentales demuestran mejoras del 40-60% en métricas "
            "clave comparados con métodos tradicionales."
        ),
    },
    "news": {
        "default": (
            "Noticias recientes reportan que {query} está generando impacto en la "
            "industria. Empresas líderes como Google, Microsoft y startups emergentes "
            "están invirtiendo significativamente en esta área. Se esperan desarrollos "
            "importantes para finales de 2026."
        ),
    },
}


def _generate_deterministic_score(query: str, source_type: str) -> float:
    """Genera un score determinístico basado en el hash del input."""
    hash_input = f"{query}:{source_type}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest()[:8], 16)
    return round(0.5 + (hash_value % 50) / 100, 2)


@task
def search_web(query: str) -> dict:
    """Busca información en la web general (mock)."""
    content = MOCK_RESULTS["web"]["default"].format(query=query)
    return {
        "source_name": "Web Search",
        "source_type": "web",
        "content": content,
        "relevance": _generate_deterministic_score(query, "web"),
    }


@task
def search_academic(query: str) -> dict:
    """Busca en fuentes académicas (mock)."""
    content = MOCK_RESULTS["academic"]["default"].format(query=query)
    return {
        "source_name": "Academic Database",
        "source_type": "academic",
        "content": content,
        "relevance": _generate_deterministic_score(query, "academic"),
    }


@task
def search_news(query: str) -> dict:
    """Busca en noticias recientes (mock)."""
    content = MOCK_RESULTS["news"]["default"].format(query=query)
    return {
        "source_name": "News Aggregator",
        "source_type": "news",
        "content": content,
        "relevance": _generate_deterministic_score(query, "news"),
    }


SEARCH_FUNCTIONS = {
    "web": search_web,
    "academic": search_academic,
    "news": search_news,
}

Cada función de búsqueda es un @task — checkpointable y ejecutable en paralelo con futures. El mock usa templates con el query interpolado para que los resultados varíen según la pregunta. El score determinístico basado en hash asegura resultados reproducibles para testing.

tools/calculator.py

"""
tools/calculator.py
Calculator tool para el AI Research Assistant.
Util para análisis que requieren cálculos numéricos.
"""

from langgraph.func import task


@task
def calculate_confidence(
    num_sources: int,
    avg_relevance: float,
    num_findings: int,
) -> float:
    """
    Calcula la confianza general del reporte basado en métricas.
    
    Factores:
    - Más fuentes = más confianza (hasta un punto)
    - Mayor relevancia promedio = más confianza
    - Más hallazgos = más confianza (hasta un punto)
    """
    source_factor = min(num_sources / 5, 1.0) * 0.4
    relevance_factor = avg_relevance * 0.4
    findings_factor = min(num_findings / 5, 1.0) * 0.2

    confidence = source_factor + relevance_factor + findings_factor
    return round(min(confidence, 1.0), 2)

El calculator es simple en v1. En módulos posteriores se extenderá para análisis más complejos (cost estimation, token counting, etc.).

Paso 4: El agente principal (agents/researcher.py)

Este es el corazón del sistema. Un @entrypoint que orquesta todo el flujo de investigación: descomponer el tema → buscar en paralelo → sintetizar → generar reporte structured.

"""
agents/researcher.py
Agente principal del AI Research Assistant (v1).
Usa Functional API de LangGraph: @entrypoint + @task.
"""

import json
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

import sys
sys.path.insert(0, ".")

from config.settings import (
    MODEL_NAME,
    MODEL_TEMPERATURE,
    MAX_SUB_QUERIES,
    SEARCH_SOURCES,
)
from state.research_state import (
    ResearchReport,
    Source,
    KeyFinding,
    SubQuery,
)
from tools.web_search import SEARCH_FUNCTIONS
from tools.calculator import calculate_confidence


model = init_chat_model(MODEL_NAME, temperature=MODEL_TEMPERATURE)


# =============================================================================
# TASK: Descomponer tema en sub-queries
# =============================================================================

@task
def decompose_query(topic: str) -> list[dict]:
    """Descompone un tema de investigación en sub-preguntas específicas."""
    response = model.invoke(
        f"Eres un investigador experto. Descompone este tema en "
        f"{MAX_SUB_QUERIES} sub-preguntas específicas e investigables.\n\n"
        f"Tema: {topic}\n\n"
        f"Responde en JSON (sin markdown, sin ```json):\n"
        f'[{{"query": "sub-pregunta", "rationale": "por qué es relevante"}}]\n\n'
        f"Solo el JSON, nada más."
    )

    try:
        queries = json.loads(response.content)
        return queries[:MAX_SUB_QUERIES]
    except json.JSONDecodeError:
        return [
            {"query": topic, "rationale": "Query original como fallback"},
            {"query": f"avances recientes en {topic}", "rationale": "Tendencias actuales"},
            {"query": f"aplicaciones prácticas de {topic}", "rationale": "Uso real"},
        ]


# =============================================================================
# TASK: Buscar en todas las fuentes para una sub-query
# =============================================================================

@task
def search_all_sources(query: str) -> list[dict]:
    """Busca en todas las fuentes configuradas para una query."""
    futures = []
    for source_type in SEARCH_SOURCES:
        search_fn = SEARCH_FUNCTIONS.get(source_type)
        if search_fn:
            futures.append(search_fn(query))

    results = [f.result() for f in futures]
    return results


# =============================================================================
# TASK: Sintetizar hallazgos en key findings
# =============================================================================

@task
def synthesize_findings(topic: str, all_results: list[dict]) -> list[dict]:
    """Sintetiza resultados de búsqueda en hallazgos clave."""
    results_text = ""
    for i, result in enumerate(all_results, 1):
        results_text += (
            f"\nFuente {i} ({result['source_type']}): {result['content']}\n"
        )

    response = model.invoke(
        f"Eres un analista de investigación. Basándote en estas fuentes, "
        f"identifica 3-5 hallazgos clave sobre '{topic}'.\n\n"
        f"Fuentes:\n{results_text}\n\n"
        f"Responde en JSON (sin markdown, sin ```json):\n"
        f'[{{"title": "título corto", "description": "descripción de 1-2 oraciones", '
        f'"confidence": 0.8}}]\n\n'
        f"confidence es de 0.0 a 1.0. Solo el JSON, nada más."
    )

    try:
        findings = json.loads(response.content)
        return findings[:5]
    except json.JSONDecodeError:
        return [{
            "title": "Hallazgo general",
            "description": f"La investigación sobre {topic} muestra resultados relevantes en múltiples fuentes.",
            "confidence": 0.6,
        }]


# =============================================================================
# TASK: Generar resumen ejecutivo
# =============================================================================

@task
def generate_summary(topic: str, findings: list[dict]) -> str:
    """Genera un resumen ejecutivo de 2-3 oraciones."""
    findings_text = "\n".join(
        f"- {f['title']}: {f['description']}" for f in findings
    )

    response = model.invoke(
        f"Genera un resumen ejecutivo de 2-3 oraciones sobre la investigación "
        f"del tema '{topic}'.\n\n"
        f"Hallazgos principales:\n{findings_text}\n\n"
        f"Solo el resumen, sin título ni formato extra."
    )
    return response.content.strip()


# =============================================================================
# ENTRYPOINT: Agente de investigación principal
# =============================================================================

memory = MemorySaver()


@entrypoint(checkpointer=memory)
def research_agent(topic: str) -> dict:
    """
    AI Research Assistant v1.
    Flujo: topic → decompose → search (parallel) → synthesize → report.
    """
    print(f"\n{'=' * 60}")
    print(f"  🔬 AI Research Assistant v1")
    print(f"  Tema: {topic}")
    print(f"{'=' * 60}")

    # --- Paso 1: Descomponer el tema ---
    print(f"\n📋 Paso 1: Descomponiendo tema en sub-queries...")
    sub_queries_raw = decompose_query(topic).result()
    sub_queries = [SubQuery(**sq) for sq in sub_queries_raw]
    print(f"   ✓ {len(sub_queries)} sub-queries generadas:")
    for i, sq in enumerate(sub_queries, 1):
        print(f"     {i}. {sq.query}")

    # --- Paso 2: Buscar en paralelo ---
    print(f"\n🔍 Paso 2: Buscando en {len(SEARCH_SOURCES)} fuentes por sub-query...")
    all_results = []
    search_futures = [
        search_all_sources(sq.query) for sq in sub_queries
    ]

    for i, future in enumerate(search_futures):
        results = future.result()
        all_results.extend(results)
        print(f"   ✓ Sub-query {i + 1}: {len(results)} fuentes consultadas")

    print(f"   Total: {len(all_results)} resultados recopilados")

    # --- Paso 3: Sintetizar hallazgos ---
    print(f"\n🧠 Paso 3: Sintetizando hallazgos...")
    findings_raw = synthesize_findings(topic, all_results).result()
    findings = [KeyFinding(**f) for f in findings_raw]
    print(f"   ✓ {len(findings)} hallazgos identificados:")
    for i, f in enumerate(findings, 1):
        print(f"     {i}. [{f.confidence:.0%}] {f.title}")

    # --- Paso 4: Generar resumen ---
    print(f"\n📝 Paso 4: Generando resumen ejecutivo...")
    summary = generate_summary(topic, findings_raw).result()
    print(f"   ✓ Resumen generado ({len(summary)} chars)")

    # --- Paso 5: Calcular confianza ---
    print(f"\n📊 Paso 5: Calculando confianza del reporte...")
    avg_relevance = (
        sum(r["relevance"] for r in all_results) / len(all_results)
        if all_results
        else 0.5
    )
    confidence = calculate_confidence(
        num_sources=len(all_results),
        avg_relevance=avg_relevance,
        num_findings=len(findings),
    ).result()
    print(f"   ✓ Confianza: {confidence:.0%}")

    # --- Paso 6: Construir reporte structured ---
    print(f"\n📄 Paso 6: Construyendo reporte...")
    sources = [
        Source(
            name=r["source_name"],
            source_type=r["source_type"],
            content=r["content"],
        )
        for r in all_results
    ]

    report = ResearchReport(
        topic=topic,
        summary=summary,
        key_findings=findings,
        sources=sources,
        sub_queries=[sq.query for sq in sub_queries],
        confidence=confidence,
    )

    print(f"   ✓ Reporte generado exitosamente")
    print(f"\n{'=' * 60}")

    return report.model_dump()

Analicemos las decisiones de diseño:

¿Por qué decompose_query pide JSON? Porque necesitamos structured output del LLM. En v1 usamos JSON parsing directo con un fallback. En módulos posteriores podrías usar with_structured_output de LangChain para mayor robustez.

¿Por qué búsqueda en paralelo con futures? Cada search_all_sources lanza las 3 búsquedas (web, academic, news) en paralelo usando futures de @task. Y las búsquedas por sub-query también se lanzan en paralelo. Esto es significativamente más rápido que secuencial.

¿Por qué Pydantic para el reporte? El output structured es fundamental. Cualquier sistema downstream (API, dashboard, base de datos) puede consumir el reporte porque tiene un schema definido. No es un string libre que necesita parsing.

¿Por qué MemorySaver? El checkpointer habilita durabilidad. Si el agente falla a mitad de ejecución, puede resumir desde el último checkpoint. En M8 lo reemplazarás con PostgresSaver para persistencia real.

Paso 5: CLI (main.py)

El CLI es la interfaz de usuario del sistema. En v1 es simple: recibe un tema por input o argumento, ejecuta el agente, y muestra el reporte formateado.

"""
main.py
CLI entrypoint para el AI Research Assistant.
"""

import sys
import json
import uuid

sys.path.insert(0, ".")

from agents.researcher import research_agent


def format_report(report: dict) -> str:
    """Formatea el reporte para display en terminal."""
    lines = []
    lines.append("")
    lines.append("╔" + "═" * 58 + "╗")
    lines.append("║" + "  📄 REPORTE DE INVESTIGACIÓN".center(58) + "║")
    lines.append("╚" + "═" * 58 + "╝")

    lines.append(f"\n📌 Tema: {report['topic']}")
    lines.append(f"📅 Generado: {report['generated_at']}")
    lines.append(f"🎯 Confianza: {report['confidence']:.0%}")

    lines.append(f"\n{'─' * 60}")
    lines.append("📋 RESUMEN EJECUTIVO")
    lines.append(f"{'─' * 60}")
    lines.append(report["summary"])

    lines.append(f"\n{'─' * 60}")
    lines.append("🔍 SUB-QUERIES INVESTIGADAS")
    lines.append(f"{'─' * 60}")
    for i, sq in enumerate(report["sub_queries"], 1):
        lines.append(f"  {i}. {sq}")

    lines.append(f"\n{'─' * 60}")
    lines.append("💡 HALLAZGOS PRINCIPALES")
    lines.append(f"{'─' * 60}")
    for i, finding in enumerate(report["key_findings"], 1):
        conf = finding["confidence"]
        lines.append(f"\n  {i}. {finding['title']} [{conf:.0%} confianza]")
        lines.append(f"     {finding['description']}")

    lines.append(f"\n{'─' * 60}")
    lines.append(f"📚 FUENTES CONSULTADAS ({len(report['sources'])})")
    lines.append(f"{'─' * 60}")
    seen = set()
    for source in report["sources"]:
        key = f"{source['name']}:{source['source_type']}"
        if key not in seen:
            seen.add(key)
            lines.append(f"  • [{source['source_type'].upper()}] {source['name']}")

    lines.append(f"\n{'═' * 60}")

    return "\n".join(lines)


def run_interactive():
    """Modo interactivo: el usuario escribe temas."""
    print("=" * 60)
    print("  🔬 AI Research Assistant v1")
    print("  Escribe un tema para investigar.")
    print("  Comandos: 'salir' para terminar")
    print("=" * 60)

    while True:
        try:
            topic = input("\n🔎 Tema: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not topic:
            continue

        if topic.lower() in ("salir", "exit", "quit"):
            print("\n¡Hasta luego!")
            break

        thread_id = f"research-{uuid.uuid4().hex[:8]}"

        try:
            report = research_agent.invoke(
                topic,
                config={"configurable": {"thread_id": thread_id}},
            )
            print(format_report(report))

        except Exception as e:
            print(f"\n❌ Error durante la investigación: {e}")
            print("   Intenta con otro tema.")


def run_single(topic: str):
    """Ejecuta una sola investigación."""
    thread_id = f"research-{uuid.uuid4().hex[:8]}"

    report = research_agent.invoke(
        topic,
        config={"configurable": {"thread_id": thread_id}},
    )
    print(format_report(report))

    print("\n📦 Reporte JSON:")
    print(json.dumps(report, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    if len(sys.argv) > 1:
        run_single(" ".join(sys.argv[1:]))
    else:
        run_interactive()

El CLI tiene dos modos:

Interactivo: python main.py — loop donde el usuario escribe temas
Single shot: python main.py "impacto de LLMs en educación" — una investigación y sale

El reporte se muestra formateado para la terminal y también como JSON crudo. El JSON es útil para verificar que el structured output es válido.

Paso 6: requirements.txt

# requirements.txt
langchain>=0.3.0
langgraph>=0.3.0
langchain-openai>=0.3.0
pydantic>=2.0.0
python-dotenv>=1.0.0

Código completo por archivo

Para referencia rápida, aquí están todos los archivos consolidados. Si seguiste los pasos anteriores, ya los tienes. Si prefieres copiar y ejecutar directamente, aquí están.

state/research_state.py

"""
state/research_state.py
Modelos de datos para el AI Research Assistant.
"""

from pydantic import BaseModel, Field
from datetime import datetime


class Source(BaseModel):
    name: str = Field(description="Nombre de la fuente")
    source_type: str = Field(description="Tipo: web, academic, news")
    content: str = Field(description="Contenido extraído de la fuente")


class KeyFinding(BaseModel):
    title: str = Field(description="Título del hallazgo")
    description: str = Field(description="Descripción del hallazgo")
    confidence: float = Field(
        description="Confianza en el hallazgo (0.0 a 1.0)",
        ge=0.0,
        le=1.0,
    )


class ResearchReport(BaseModel):
    topic: str = Field(description="Tema investigado")
    summary: str = Field(description="Resumen ejecutivo (2-3 oraciones)")
    key_findings: list[KeyFinding] = Field(
        description="Hallazgos principales",
        min_length=1,
    )
    sources: list[Source] = Field(
        description="Fuentes consultadas",
        min_length=1,
    )
    sub_queries: list[str] = Field(
        description="Sub-preguntas generadas para la investigación",
    )
    confidence: float = Field(
        description="Confianza general del reporte (0.0 a 1.0)",
        ge=0.0,
        le=1.0,
    )
    generated_at: str = Field(
        default_factory=lambda: datetime.now().isoformat(),
        description="Timestamp de generación",
    )


class SubQuery(BaseModel):
    query: str = Field(description="La sub-pregunta")
    rationale: str = Field(description="Por qué esta pregunta es relevante")

config/settings.py

"""
config/settings.py
Configuración del AI Research Assistant.
"""

from dotenv import load_dotenv
load_dotenv()

MODEL_NAME = "openai:gpt-4.1-mini"
MODEL_TEMPERATURE = 0.2

MAX_SUB_QUERIES = 4
MAX_SOURCES_PER_QUERY = 3

SEARCH_SOURCES = ["web", "academic", "news"]

tools/web_search.py

"""
tools/web_search.py
Mock web search tool para el AI Research Assistant.
"""

import hashlib
from langgraph.func import task


MOCK_RESULTS = {
    "web": {
        "default": (
            "Múltiples fuentes web coinciden en que {query} es un tema de creciente "
            "interés. Expertos destacan avances significativos en los últimos 2 años. "
            "Las aplicaciones prácticas incluyen automatización, análisis de datos "
            "y generación de contenido."
        ),
    },
    "academic": {
        "default": (
            "Investigación reciente (2025-2026) muestra que {query} tiene fundamentos "
            "teóricos sólidos respaldados por múltiples estudios peer-reviewed. "
            "Los resultados experimentales demuestran mejoras del 40-60% en métricas "
            "clave comparados con métodos tradicionales."
        ),
    },
    "news": {
        "default": (
            "Noticias recientes reportan que {query} está generando impacto en la "
            "industria. Empresas líderes como Google, Microsoft y startups emergentes "
            "están invirtiendo significativamente en esta área. Se esperan desarrollos "
            "importantes para finales de 2026."
        ),
    },
}


def _generate_deterministic_score(query: str, source_type: str) -> float:
    hash_input = f"{query}:{source_type}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest()[:8], 16)
    return round(0.5 + (hash_value % 50) / 100, 2)


@task
def search_web(query: str) -> dict:
    """Busca información en la web general (mock)."""
    content = MOCK_RESULTS["web"]["default"].format(query=query)
    return {
        "source_name": "Web Search",
        "source_type": "web",
        "content": content,
        "relevance": _generate_deterministic_score(query, "web"),
    }


@task
def search_academic(query: str) -> dict:
    """Busca en fuentes académicas (mock)."""
    content = MOCK_RESULTS["academic"]["default"].format(query=query)
    return {
        "source_name": "Academic Database",
        "source_type": "academic",
        "content": content,
        "relevance": _generate_deterministic_score(query, "academic"),
    }


@task
def search_news(query: str) -> dict:
    """Busca en noticias recientes (mock)."""
    content = MOCK_RESULTS["news"]["default"].format(query=query)
    return {
        "source_name": "News Aggregator",
        "source_type": "news",
        "content": content,
        "relevance": _generate_deterministic_score(query, "news"),
    }


SEARCH_FUNCTIONS = {
    "web": search_web,
    "academic": search_academic,
    "news": search_news,
}

tools/calculator.py

"""
tools/calculator.py
Calculator tool para el AI Research Assistant.
"""

from langgraph.func import task


@task
def calculate_confidence(
    num_sources: int,
    avg_relevance: float,
    num_findings: int,
) -> float:
    source_factor = min(num_sources / 5, 1.0) * 0.4
    relevance_factor = avg_relevance * 0.4
    findings_factor = min(num_findings / 5, 1.0) * 0.2

    confidence = source_factor + relevance_factor + findings_factor
    return round(min(confidence, 1.0), 2)

agents/researcher.py

"""
agents/researcher.py
Agente principal del AI Research Assistant (v1).
"""

import json
from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

import sys
sys.path.insert(0, ".")

from config.settings import (
    MODEL_NAME,
    MODEL_TEMPERATURE,
    MAX_SUB_QUERIES,
    SEARCH_SOURCES,
)
from state.research_state import (
    ResearchReport,
    Source,
    KeyFinding,
    SubQuery,
)
from tools.web_search import SEARCH_FUNCTIONS
from tools.calculator import calculate_confidence


model = init_chat_model(MODEL_NAME, temperature=MODEL_TEMPERATURE)


@task
def decompose_query(topic: str) -> list[dict]:
    """Descompone un tema de investigación en sub-preguntas específicas."""
    response = model.invoke(
        f"Eres un investigador experto. Descompone este tema en "
        f"{MAX_SUB_QUERIES} sub-preguntas específicas e investigables.\n\n"
        f"Tema: {topic}\n\n"
        f"Responde en JSON (sin markdown, sin ```json):\n"
        f'[{{"query": "sub-pregunta", "rationale": "por qué es relevante"}}]\n\n'
        f"Solo el JSON, nada más."
    )
    try:
        queries = json.loads(response.content)
        return queries[:MAX_SUB_QUERIES]
    except json.JSONDecodeError:
        return [
            {"query": topic, "rationale": "Query original como fallback"},
            {"query": f"avances recientes en {topic}", "rationale": "Tendencias actuales"},
            {"query": f"aplicaciones prácticas de {topic}", "rationale": "Uso real"},
        ]


@task
def search_all_sources(query: str) -> list[dict]:
    """Busca en todas las fuentes configuradas para una query."""
    futures = []
    for source_type in SEARCH_SOURCES:
        search_fn = SEARCH_FUNCTIONS.get(source_type)
        if search_fn:
            futures.append(search_fn(query))
    results = [f.result() for f in futures]
    return results


@task
def synthesize_findings(topic: str, all_results: list[dict]) -> list[dict]:
    """Sintetiza resultados de búsqueda en hallazgos clave."""
    results_text = ""
    for i, result in enumerate(all_results, 1):
        results_text += (
            f"\nFuente {i} ({result['source_type']}): {result['content']}\n"
        )

    response = model.invoke(
        f"Eres un analista de investigación. Basándote en estas fuentes, "
        f"identifica 3-5 hallazgos clave sobre '{topic}'.\n\n"
        f"Fuentes:\n{results_text}\n\n"
        f"Responde en JSON (sin markdown, sin ```json):\n"
        f'[{{"title": "título corto", "description": "descripción de 1-2 oraciones", '
        f'"confidence": 0.8}}]\n\n'
        f"confidence es de 0.0 a 1.0. Solo el JSON, nada más."
    )
    try:
        findings = json.loads(response.content)
        return findings[:5]
    except json.JSONDecodeError:
        return [{
            "title": "Hallazgo general",
            "description": f"La investigación sobre {topic} muestra resultados relevantes.",
            "confidence": 0.6,
        }]


@task
def generate_summary(topic: str, findings: list[dict]) -> str:
    """Genera un resumen ejecutivo de 2-3 oraciones."""
    findings_text = "\n".join(
        f"- {f['title']}: {f['description']}" for f in findings
    )
    response = model.invoke(
        f"Genera un resumen ejecutivo de 2-3 oraciones sobre la investigación "
        f"del tema '{topic}'.\n\n"
        f"Hallazgos principales:\n{findings_text}\n\n"
        f"Solo el resumen, sin título ni formato extra."
    )
    return response.content.strip()


memory = MemorySaver()


@entrypoint(checkpointer=memory)
def research_agent(topic: str) -> dict:
    """
    AI Research Assistant v1.
    Flujo: topic → decompose → search (parallel) → synthesize → report.
    """
    print(f"\n{'=' * 60}")
    print(f"  🔬 AI Research Assistant v1")
    print(f"  Tema: {topic}")
    print(f"{'=' * 60}")

    # Paso 1: Descomponer
    print(f"\n📋 Paso 1: Descomponiendo tema en sub-queries...")
    sub_queries_raw = decompose_query(topic).result()
    sub_queries = [SubQuery(**sq) for sq in sub_queries_raw]
    print(f"   ✓ {len(sub_queries)} sub-queries generadas:")
    for i, sq in enumerate(sub_queries, 1):
        print(f"     {i}. {sq.query}")

    # Paso 2: Buscar en paralelo
    print(f"\n🔍 Paso 2: Buscando en {len(SEARCH_SOURCES)} fuentes por sub-query...")
    all_results = []
    search_futures = [search_all_sources(sq.query) for sq in sub_queries]

    for i, future in enumerate(search_futures):
        results = future.result()
        all_results.extend(results)
        print(f"   ✓ Sub-query {i + 1}: {len(results)} fuentes consultadas")

    print(f"   Total: {len(all_results)} resultados recopilados")

    # Paso 3: Sintetizar
    print(f"\n🧠 Paso 3: Sintetizando hallazgos...")
    findings_raw = synthesize_findings(topic, all_results).result()
    findings = [KeyFinding(**f) for f in findings_raw]
    print(f"   ✓ {len(findings)} hallazgos identificados:")
    for i, f in enumerate(findings, 1):
        print(f"     {i}. [{f.confidence:.0%}] {f.title}")

    # Paso 4: Resumen
    print(f"\n📝 Paso 4: Generando resumen ejecutivo...")
    summary = generate_summary(topic, findings_raw).result()
    print(f"   ✓ Resumen generado ({len(summary)} chars)")

    # Paso 5: Confianza
    print(f"\n📊 Paso 5: Calculando confianza del reporte...")
    avg_relevance = (
        sum(r["relevance"] for r in all_results) / len(all_results)
        if all_results else 0.5
    )
    confidence = calculate_confidence(
        num_sources=len(all_results),
        avg_relevance=avg_relevance,
        num_findings=len(findings),
    ).result()
    print(f"   ✓ Confianza: {confidence:.0%}")

    # Paso 6: Reporte
    print(f"\n📄 Paso 6: Construyendo reporte...")
    sources = [
        Source(
            name=r["source_name"],
            source_type=r["source_type"],
            content=r["content"],
        )
        for r in all_results
    ]

    report = ResearchReport(
        topic=topic,
        summary=summary,
        key_findings=findings,
        sources=sources,
        sub_queries=[sq.query for sq in sub_queries],
        confidence=confidence,
    )

    print(f"   ✓ Reporte generado exitosamente")
    print(f"\n{'=' * 60}")

    return report.model_dump()

main.py

"""
main.py
CLI entrypoint para el AI Research Assistant.
"""

import sys
import json
import uuid

sys.path.insert(0, ".")

from agents.researcher import research_agent


def format_report(report: dict) -> str:
    lines = []
    lines.append("")
    lines.append("╔" + "═" * 58 + "╗")
    lines.append("║" + "  📄 REPORTE DE INVESTIGACIÓN".center(58) + "║")
    lines.append("╚" + "═" * 58 + "╝")

    lines.append(f"\n📌 Tema: {report['topic']}")
    lines.append(f"📅 Generado: {report['generated_at']}")
    lines.append(f"🎯 Confianza: {report['confidence']:.0%}")

    lines.append(f"\n{'─' * 60}")
    lines.append("📋 RESUMEN EJECUTIVO")
    lines.append(f"{'─' * 60}")
    lines.append(report["summary"])

    lines.append(f"\n{'─' * 60}")
    lines.append("🔍 SUB-QUERIES INVESTIGADAS")
    lines.append(f"{'─' * 60}")
    for i, sq in enumerate(report["sub_queries"], 1):
        lines.append(f"  {i}. {sq}")

    lines.append(f"\n{'─' * 60}")
    lines.append("💡 HALLAZGOS PRINCIPALES")
    lines.append(f"{'─' * 60}")
    for i, finding in enumerate(report["key_findings"], 1):
        conf = finding["confidence"]
        lines.append(f"\n  {i}. {finding['title']} [{conf:.0%} confianza]")
        lines.append(f"     {finding['description']}")

    lines.append(f"\n{'─' * 60}")
    lines.append(f"📚 FUENTES CONSULTADAS ({len(report['sources'])})")
    lines.append(f"{'─' * 60}")
    seen = set()
    for source in report["sources"]:
        key = f"{source['name']}:{source['source_type']}"
        if key not in seen:
            seen.add(key)
            lines.append(f"  • [{source['source_type'].upper()}] {source['name']}")

    lines.append(f"\n{'═' * 60}")

    return "\n".join(lines)


def run_interactive():
    print("=" * 60)
    print("  🔬 AI Research Assistant v1")
    print("  Escribe un tema para investigar.")
    print("  Comandos: 'salir' para terminar")
    print("=" * 60)

    while True:
        try:
            topic = input("\n🔎 Tema: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\n¡Hasta luego!")
            break

        if not topic:
            continue

        if topic.lower() in ("salir", "exit", "quit"):
            print("\n¡Hasta luego!")
            break

        thread_id = f"research-{uuid.uuid4().hex[:8]}"

        try:
            report = research_agent.invoke(
                topic,
                config={"configurable": {"thread_id": thread_id}},
            )
            print(format_report(report))

        except Exception as e:
            print(f"\n❌ Error durante la investigación: {e}")
            print("   Intenta con otro tema.")


def run_single(topic: str):
    thread_id = f"research-{uuid.uuid4().hex[:8]}"

    report = research_agent.invoke(
        topic,
        config={"configurable": {"thread_id": thread_id}},
    )
    print(format_report(report))

    print("\n📦 Reporte JSON:")
    print(json.dumps(report, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    if len(sys.argv) > 1:
        run_single(" ".join(sys.argv[1:]))
    else:
        run_interactive()

Ejecución

Modo single (una investigación)

cd research-assistant
python main.py "impacto de la inteligencia artificial en la educación"

Modo interactivo

cd research-assistant
python main.py

Criterios de éxito

Tu proyecto está completo cuando cumples estos criterios:

✅ El agente descompone el tema en 3+ sub-queries — el LLM genera sub-preguntas específicas y relevantes a partir del tema dado
✅ Las búsquedas se ejecutan en paralelo — las 3 fuentes (web, academic, news) se buscan simultáneamente para cada sub-query usando el pattern de futures
✅ El reporte final es structured (Pydantic) — el output es un ResearchReport válido con todos los campos requeridos: topic, summary, key_findings, sources, sub_queries, confidence
✅ El streaming muestra progreso — durante la ejecución ves cada paso completarse: descomposición, búsqueda, síntesis, resumen, confianza, reporte
✅ El CLI funciona en ambos modos — interactivo (sin argumentos) y single-shot (con argumento)
✅ El JSON del reporte es válido — puedes copiar el output JSON y parsearlo sin errores

Escenarios de prueba

Test 1: Tema técnico

🔎 Tema: machine learning aplicado a diagnóstico médico

============================================================
  🔬 AI Research Assistant v1
  Tema: machine learning aplicado a diagnóstico médico
============================================================

📋 Paso 1: Descomponiendo tema en sub-queries...
   ✓ 4 sub-queries generadas:
     1. ¿Qué algoritmos de ML se usan más en diagnóstico médico?
     2. ¿Cuál es la precisión actual de los modelos de ML en detección de enfermedades?
     3. ¿Qué desafíos éticos plantea el uso de ML en medicina?
     4. ¿Qué hospitales o instituciones están implementando ML en diagnóstico?

🔍 Paso 2: Buscando en 3 fuentes por sub-query...
   ✓ Sub-query 1: 3 fuentes consultadas
   ✓ Sub-query 2: 3 fuentes consultadas
   ✓ Sub-query 3: 3 fuentes consultadas
   ✓ Sub-query 4: 3 fuentes consultadas
   Total: 12 resultados recopilados

🧠 Paso 3: Sintetizando hallazgos...
   ✓ 4 hallazgos identificados:
     1. [85%] Algoritmos predominantes en diagnóstico
     2. [80%] Precisión comparable a especialistas
     3. [75%] Desafíos éticos significativos
     4. [70%] Adopción institucional creciente

📝 Paso 4: Generando resumen ejecutivo...
   ✓ Resumen generado (180 chars)

📊 Paso 5: Calculando confianza del reporte...
   ✓ Confianza: 82%

📄 Paso 6: Construyendo reporte...
   ✓ Reporte generado exitosamente

============================================================

╔══════════════════════════════════════════════════════════╗
║            📄 REPORTE DE INVESTIGACIÓN                  ║
╚══════════════════════════════════════════════════════════╝

📌 Tema: machine learning aplicado a diagnóstico médico
📅 Generado: 2026-03-08T...
🎯 Confianza: 82%

────────────────────────────────────────────────────────────
📋 RESUMEN EJECUTIVO
────────────────────────────────────────────────────────────
La aplicación de machine learning en diagnóstico médico muestra
avances significativos con precisión comparable a especialistas...

────────────────────────────────────────────────────────────
💡 HALLAZGOS PRINCIPALES
────────────────────────────────────────────────────────────

  1. Algoritmos predominantes en diagnóstico [85% confianza]
     Deep learning y redes neuronales convolucionales lideran...

Test 2: Tema general

🔎 Tema: tendencias en trabajo remoto 2026

[El agente genera sub-queries como: regulaciones laborales,
herramientas de productividad, impacto en salud mental,
modelos híbridos, etc. El reporte sintetiza fuentes web,
académicas y de noticias.]

Test 3: Tema corto y vago

🔎 Tema: Python

[El agente descompone "Python" en sub-queries más específicas:
historia y evolución, aplicaciones principales, comparación
con otros lenguajes, ecosistema de librerías. Demuestra que
el agente agrega valor incluso con input vago.]

Test 4: Verificar JSON structured

python main.py "energías renovables" 2>/dev/null | grep -A 999 "Reporte JSON" | python -m json.tool

Si el JSON se parsea sin errores, el structured output es válido.

Errores comunes

1. `ModuleNotFoundError: No module named 'config'` o `'state'` o `'tools'`

Causa: Python no encuentra los módulos porque no estás ejecutando desde el directorio correcto.

Solución: Ejecuta siempre desde la raíz del proyecto:

cd research-assistant
python main.py

El sys.path.insert(0, ".") en los archivos asegura que Python busque módulos en el directorio actual. Si ejecutas desde otro directorio, no los encontrará.

2. `json.JSONDecodeError` al parsear respuesta del LLM

Causa: El LLM a veces envuelve el JSON en markdown (```json ... ```) o agrega texto antes/después. Esto es especialmente común con modelos más pequeños.

Solución: El código ya tiene fallbacks para este caso. Si ocurre frecuentemente, puedes agregar un paso de limpieza:

import re

def clean_json(text: str) -> str:
    """Extrae JSON de una respuesta que puede incluir markdown."""
    match = re.search(r'\[.*\]', text, re.DOTALL)
    if match:
        return match.group(0)
    return text

3. El agente genera solo 1-2 sub-queries en vez de 4

Causa: El LLM puede generar menos queries de las pedidas si considera que el tema es simple, o si el JSON parsing trunca resultados.

Solución: El fallback en decompose_query genera 3 queries por defecto si el parsing falla. Si necesitas exactamente N queries, puedes agregar validación:

while len(queries) < MAX_SUB_QUERIES:
    queries.append({
        "query": f"aspecto adicional de {topic}",
        "rationale": "Completar mínimo de sub-queries",
    })

4. `ValidationError` de Pydantic al construir el reporte

Causa: Algún campo del reporte no cumple las validaciones (ej: confidence fuera del rango 0-1, key_findings vacío, etc.).

Solución: Revisa qué campo falla en el mensaje de error. Los errores más comunes:

# confidence fuera de rango
confidence = max(0.0, min(1.0, calculated_value))

# key_findings vacío
if not findings:
    findings = [KeyFinding(
        title="Sin hallazgos específicos",
        description="No se identificaron hallazgos con la información disponible.",
        confidence=0.3,
    )]

5. Las búsquedas no se ejecutan en paralelo

Causa: Si el checkpointer no está activo o hay un error en la configuración de futures, las tareas se ejecutan secuencialmente.

Solución: Verifica que estás usando el patrón correcto de futures — lanzar todas las tareas primero, y luego recoger resultados:

# ✅ Correcto: lanzar todas primero
futures = [search_all_sources(sq.query) for sq in sub_queries]
results = [f.result() for f in futures]

# ❌ Incorrecto: resultado inmediato (secuencial)
results = [search_all_sources(sq.query).result() for sq in sub_queries]

6. `OPENAI_API_KEY` no encontrada

Causa: El archivo .env no existe, no está en el directorio correcto, o la variable tiene un nombre diferente.

Solución: Verifica:

# Verificar que .env existe en el directorio del proyecto
ls -la research-assistant/.env

# Verificar contenido (sin revelar la key completa)
head -c 20 research-assistant/.env

7. El reporte siempre tiene la misma confianza

Causa: La función calculate_confidence usa métricas fijas (número de fuentes, relevancia promedio, número de hallazgos). Con búsquedas mock, estas métricas varían poco.

Solución: Esto es esperado en v1 con mocks. Cuando en M7 reemplaces los mocks con búsqueda real, las métricas variarán naturalmente según la calidad real de los resultados.

8. `TypeError: 'Future' object is not subscriptable`

Causa: Estás intentando acceder a un campo del resultado antes de llamar .result():

# ❌ Error: future no es un dict
result = search_web(query)
content = result["content"]

# ✅ Correcto: primero .result()
result = search_web(query).result()
content = result["content"]

Lo que viene: la evolución del Research Assistant

Lo que construiste hoy es v1. Funciona end-to-end, pero tiene limitaciones evidentes. Cada módulo siguiente resuelve una:

Módulo	Limitación actual	Qué se agrega
M7: Flujos Avanzados	Las búsquedas mock nunca fallan	Retry logic con backoff, error handling robusto, búsqueda real con fallback
M8: Memoria	Cada investigación empieza de cero	Persistencia con PostgresSaver, resumir investigaciones largas, memoria de preferencias del usuario
M9: Human-in-the-Loop	El agente actúa sin supervisión	Aprobación antes de acciones costosas, review humano de síntesis, editable state
M10: Multi-Agente	Un solo agente hace todo	4 agentes especializados (researcher, analyst, writer, supervisor) coordinados
M11: Deep Agents	Planning manual en el código	Planning autónomo con `write_todos`, filesystem para almacenar investigación, subagent spawning
M12: Producción	Sin observabilidad	LangSmith tracing, evaluation datasets, token tracking, rate limiting

La v1 tiene la arquitectura correcta para soportar esta evolución. Los modelos Pydantic se extenderán. Los @task se reemplazarán con versiones más robustas. El @entrypoint eventualmente se convertirá en un sistema híbrido con StateGraph para los pasos que lo necesiten. La estructura de archivos (agents/, tools/, state/, config/) ya anticipa todo lo que viene.

No borres lo que construiste hoy. Iterarás sobre esto durante 6 módulos más.

Recursos del proyecto

LangGraph Functional API Guide — Documentación oficial de @entrypoint y @task
LangGraph Checkpointing — Cómo funciona MemorySaver y checkpointing
Pydantic v2 Documentation — Modelos de datos, validación, serialización
LangChain init_chat_model — Inicialización de modelos
LangGraph How-To Guides — Patterns prácticos para workflows
Python asyncio and Futures — Conceptos de futures en Python (análogos a los futures de LangGraph)

Módulo 6 — LangChain & LangGraph: From Chains to Agents

← AnteriorCombinar Graph y Functional API Siguiente →Introducción: Workflows Complejos del Mundo Real

Proyecto Evolutivo: Research Agent Base (v1)

Descripción del proyecto

Objetivo del proyecto

Especificaciones técnicas

Stack tecnológico

Setup inicial

Estructura del proyecto

Paso 1: Modelos de datos (state/research_state.py)

Paso 2: Configuración (config/settings.py)

Paso 3: Tools — búsqueda y cálculo (tools/)

tools/web_search.py

tools/calculator.py

Paso 4: El agente principal (agents/researcher.py)

Paso 5: CLI (main.py)

Paso 6: requirements.txt

Código completo por archivo

state/research_state.py

config/settings.py

tools/web_search.py

tools/calculator.py

agents/researcher.py

main.py

Ejecución

Modo single (una investigación)

Modo interactivo

Criterios de éxito

Escenarios de prueba

Test 1: Tema técnico

Test 2: Tema general

Test 3: Tema corto y vago

Test 4: Verificar JSON structured

Errores comunes

1. ModuleNotFoundError: No module named 'config' o 'state' o 'tools'

2. json.JSONDecodeError al parsear respuesta del LLM

3. El agente genera solo 1-2 sub-queries en vez de 4

4. ValidationError de Pydantic al construir el reporte

5. Las búsquedas no se ejecutan en paralelo

6. OPENAI_API_KEY no encontrada

7. El reporte siempre tiene la misma confianza

8. TypeError: 'Future' object is not subscriptable

Lo que viene: la evolución del Research Assistant

Recursos del proyecto

1. `ModuleNotFoundError: No module named 'config'` o `'state'` o `'tools'`

2. `json.JSONDecodeError` al parsear respuesta del LLM

4. `ValidationError` de Pydantic al construir el reporte

6. `OPENAI_API_KEY` no encontrada

8. `TypeError: 'Future' object is not subscriptable`