Module 6: Functional API

Evolving Project: Research Agent Baseline (v1)

Project overview

In the previous seven capsules you learned LangGraph's Functional API: @entrypoint to define workflows as functions, @task for checkpointable tasks, native control flow with loops and conditionals, the deep comparison between the Graph API and the Functional API, advanced patterns like futures for parallel execution, and how to combine both APIs when the case calls for it. You saw each concept on its own. Now you're going to combine everything into a real system.

But this isn't just one more mini-project. This is the start of the AI Research Assistant — a project you'll build, iterate on, and scale across the next 6 modules. What you build today is v1: a working agent that takes a research topic, breaks it into sub-questions, searches multiple sources in parallel, and generates a structured report. It's simple, but it works end-to-end. And every piece is designed to evolve.

In Module 7, you'll add retry logic and robust error handling. In Module 8, persistent memory so it remembers past research. In Module 9, human approvals before expensive actions. In Module 10, multiple specialized agents coordinated by a supervisor. In Module 11, autonomous planning with Deep Agents. In Module 12, full observability with LangSmith. Each module adds a layer on top of what you build today.

Build v1 well. It's the foundation everything else rests on.

Project goal

Build a working research agent with LangGraph's Functional API that takes a topic, breaks it into sub-queries, runs parallel searches across multiple sources, and generates a structured report with Pydantic.

By the end of this project:

🔧 You'll know how to design data models with Pydantic for structured agent output
🔧 You'll implement @task functions for decomposition, search and synthesis
🔧 You'll create an @entrypoint that orchestrates the full research flow
🔧 You'll use the Futures pattern for parallel search across multiple sources
🔧 You'll stream progress during execution
🔧 You'll build a simple CLI to interact with the agent

Technical specs

Tech stack

Component	Version	Purpose
Python	3.11+	Runtime
LangChain	v1.2+	LLM framework
LangGraph	v1.0+	Functional API (@entrypoint, @task)
langchain-openai	latest	Model provider
pydantic	v2+	Structured data models
python-dotenv	latest	Environment variables

Initial setup

pip install langchain langgraph langchain-openai pydantic python-dotenv

Create a .env file at the root of your project:

# .env
OPENAI_API_KEY=sk-...

Project structure

research-assistant/
├── .env                        # API key
├── requirements.txt            # Dependencies
├── agents/
│   └── researcher.py           # @entrypoint — main agent
├── tools/
│   ├── web_search.py           # Mock web search (@task)
│   └── calculator.py           # Calculator tool (@task)
├── state/
│   └── research_state.py       # Pydantic models for the report
├── config/
│   └── settings.py             # Project configuration
└── main.py                     # CLI entrypoint

This structure looks like overkill for a v1 project. It isn't. Every directory has a purpose that will reveal itself in later modules:

agents/ — In M10 you'll have analyst.py, writer.py, supervisor.py here
tools/ — In M7 you'll add document_reader.py with retry logic
state/ — In M8 the models get extended with memory fields
config/ — In M12 you'll have LangSmith configuration and rate limiting

Step 1: Data models (state/research_state.py)

The research report needs structure. It isn't a free-form string — it's an object with typed fields that any downstream system can consume. Pydantic gives you validation, serialization, and automatic documentation.

"""
state/research_state.py
Data models for the AI Research Assistant.
"""

from pydantic import BaseModel, Field
from datetime import datetime


class Source(BaseModel):
    """A source of information found during the research."""
    name: str = Field(description="Name of the source")
    source_type: str = Field(description="Type: web, academic, news")
    content: str = Field(description="Content extracted from the source")


class KeyFinding(BaseModel):
    """A key finding identified during the research."""
    title: str = Field(description="Title of the finding")
    description: str = Field(description="Description of the finding")
    confidence: float = Field(
        description="Confidence in the finding (0.0 to 1.0)",
        ge=0.0,
        le=1.0,
    )


class ResearchReport(BaseModel):
    """Structured research report."""
    topic: str = Field(description="Topic researched")
    summary: str = Field(description="Executive summary (2-3 sentences)")
    key_findings: list[KeyFinding] = Field(
        description="Main findings",
        min_length=1,
    )
    sources: list[Source] = Field(
        description="Sources consulted",
        min_length=1,
    )
    sub_queries: list[str] = Field(
        description="Sub-questions generated for the research",
    )
    confidence: float = Field(
        description="Overall confidence of the report (0.0 to 1.0)",
        ge=0.0,
        le=1.0,
    )
    generated_at: str = Field(
        default_factory=lambda: datetime.now().isoformat(),
        description="Generation timestamp",
    )


class SubQuery(BaseModel):
    """A sub-question generated from the main topic."""
    query: str = Field(description="The sub-question")
    rationale: str = Field(description="Why this question is relevant")

Each model has a clear purpose. ResearchReport is the agent's final output. Source and KeyFinding are the report's building blocks. SubQuery structures the topic decomposition.

Step 2: Configuration (config/settings.py)

"""
config/settings.py
Configuration for the 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"]

Centralizing configuration makes changes easy. When you add LangSmith configuration and rate limits in M12, everything will be in one place.

Step 3: Tools — search and calculation (tools/)

The agent's tools are @task functions. In v1 we use mock search — we don't depend on external APIs that can fail, change, or cost money. In M7 you'll add real search with retry logic.

tools/web_search.py

"""
tools/web_search.py
Mock web search tool for the AI Research Assistant.
In M7 it gets replaced with real search + retry logic.
"""

import hashlib
from langgraph.func import task


MOCK_RESULTS = {
    "web": {
        "default": (
            "Multiple web sources agree that {query} is a topic of growing "
            "interest. Experts highlight significant advances over the last 2 years. "
            "Practical applications include automation, data analysis "
            "and content generation."
        ),
    },
    "academic": {
        "default": (
            "Recent research (2025-2026) shows that {query} has solid theoretical "
            "foundations backed by multiple peer-reviewed studies. "
            "Experimental results demonstrate 40-60% improvements on key "
            "metrics compared to traditional methods."
        ),
    },
    "news": {
        "default": (
            "Recent news reports that {query} is making an impact across the "
            "industry. Leading companies like Google, Microsoft and emerging startups "
            "are investing heavily in this area. Major developments are expected "
            "by the end of 2026."
        ),
    },
}


def _generate_deterministic_score(query: str, source_type: str) -> float:
    """Generate a deterministic score based on the hash of the 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:
    """Search the general web (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:
    """Search academic sources (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:
    """Search recent news (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,
}

Each search function is a @task — checkpointable and runnable in parallel with futures. The mock uses templates with the query interpolated so the results vary with the question. The deterministic hash-based score guarantees reproducible results for testing.

tools/calculator.py

"""
tools/calculator.py
Calculator tool for the AI Research Assistant.
Useful for analyses that require numeric calculations.
"""

from langgraph.func import task


@task
def calculate_confidence(
    num_sources: int,
    avg_relevance: float,
    num_findings: int,
) -> float:
    """
    Compute the report's overall confidence based on metrics.
    
    Factors:
    - More sources = more confidence (up to a point)
    - Higher average relevance = more confidence
    - More findings = more confidence (up to a point)
    """
    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)

The calculator is simple in v1. In later modules it gets extended for more complex analysis (cost estimation, token counting, etc.).

Step 4: The main agent (agents/researcher.py)

This is the heart of the system. An @entrypoint that orchestrates the whole research flow: decompose the topic → search in parallel → synthesize → generate a structured report.

"""
agents/researcher.py
Main agent of the AI Research Assistant (v1).
Uses LangGraph's Functional API: @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: Decompose the topic into sub-queries
# =============================================================================

@task
def decompose_query(topic: str) -> list[dict]:
    """Break a research topic down into specific sub-questions."""
    response = model.invoke(
        f"You are an expert researcher. Break this topic down into "
        f"{MAX_SUB_QUERIES} specific, researchable sub-questions.\n\n"
        f"Topic: {topic}\n\n"
        f"Answer in JSON (no markdown, no ```json):\n"
        f'[{{"query": "sub-question", "rationale": "why it is relevant"}}]\n\n'
        f"Only the JSON, nothing else."
    )

    try:
        queries = json.loads(response.content)
        return queries[:MAX_SUB_QUERIES]
    except json.JSONDecodeError:
        return [
            {"query": topic, "rationale": "Original query as a fallback"},
            {"query": f"recent advances in {topic}", "rationale": "Current trends"},
            {"query": f"practical applications of {topic}", "rationale": "Real-world use"},
        ]


# =============================================================================
# TASK: Search every source for one sub-query
# =============================================================================

@task
def search_all_sources(query: str) -> list[dict]:
    """Search every configured source for one 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: Synthesize the results into key findings
# =============================================================================

@task
def synthesize_findings(topic: str, all_results: list[dict]) -> list[dict]:
    """Synthesize search results into key findings."""
    results_text = ""
    for i, result in enumerate(all_results, 1):
        results_text += (
            f"\nSource {i} ({result['source_type']}): {result['content']}\n"
        )

    response = model.invoke(
        f"You are a research analyst. Based on these sources, "
        f"identify 3-5 key findings about '{topic}'.\n\n"
        f"Sources:\n{results_text}\n\n"
        f"Answer in JSON (no markdown, no ```json):\n"
        f'[{{"title": "short title", "description": "1-2 sentence description", '
        f'"confidence": 0.8}}]\n\n'
        f"confidence goes from 0.0 to 1.0. Only the JSON, nothing else."
    )

    try:
        findings = json.loads(response.content)
        return findings[:5]
    except json.JSONDecodeError:
        return [{
            "title": "General finding",
            "description": f"Research on {topic} shows relevant results across multiple sources.",
            "confidence": 0.6,
        }]


# =============================================================================
# TASK: Generate the executive summary
# =============================================================================

@task
def generate_summary(topic: str, findings: list[dict]) -> str:
    """Generate a 2-3 sentence executive summary."""
    findings_text = "\n".join(
        f"- {f['title']}: {f['description']}" for f in findings
    )

    response = model.invoke(
        f"Generate a 2-3 sentence executive summary of the research "
        f"on the topic '{topic}'.\n\n"
        f"Main findings:\n{findings_text}\n\n"
        f"Only the summary, no title and no extra formatting."
    )
    return response.content.strip()


# =============================================================================
# ENTRYPOINT: Main research agent
# =============================================================================

memory = MemorySaver()


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

    # --- Step 1: Decompose the topic ---
    print(f"\n📋 Step 1: Decomposing topic into 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 generated:")
    for i, sq in enumerate(sub_queries, 1):
        print(f"     {i}. {sq.query}")

    # --- Step 2: Search in parallel ---
    print(f"\n🔍 Step 2: Searching {len(SEARCH_SOURCES)} sources per 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)} sources consulted")

    print(f"   Total: {len(all_results)} results collected")

    # --- Step 3: Synthesize the findings ---
    print(f"\n🧠 Step 3: Synthesizing findings...")
    findings_raw = synthesize_findings(topic, all_results).result()
    findings = [KeyFinding(**f) for f in findings_raw]
    print(f"   ✓ {len(findings)} findings identified:")
    for i, f in enumerate(findings, 1):
        print(f"     {i}. [{f.confidence:.0%}] {f.title}")

    # --- Step 4: Generate the summary ---
    print(f"\n📝 Step 4: Generating executive summary...")
    summary = generate_summary(topic, findings_raw).result()
    print(f"   ✓ Summary generated ({len(summary)} chars)")

    # --- Step 5: Compute confidence ---
    print(f"\n📊 Step 5: Computing the report's confidence...")
    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"   ✓ Confidence: {confidence:.0%}")

    # --- Step 6: Build the structured report ---
    print(f"\n📄 Step 6: Building the report...")
    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"   ✓ Report generated successfully")
    print(f"\n{'=' * 60}")

    return report.model_dump()

Let's go through the design decisions:

Why does decompose_query ask for JSON? Because we need structured output from the LLM. In v1 we use direct JSON parsing with a fallback. In later modules you could use LangChain's with_structured_output for more robustness.

Why parallel search with futures? Each search_all_sources launches the 3 searches (web, academic, news) in parallel using @task futures. And the searches for each sub-query are launched in parallel too. This is significantly faster than sequential.

Why Pydantic for the report? Structured output is fundamental. Any downstream system (an API, a dashboard, a database) can consume the report because it has a defined schema. It isn't a free-form string that needs parsing.

Why MemorySaver? The checkpointer enables durability. If the agent fails mid-execution, it can resume from the last checkpoint. In M8 you'll replace it with PostgresSaver for real persistence.

Step 5: CLI (main.py)

The CLI is the system's user interface. In v1 it's simple: it takes a topic from input or an argument, runs the agent, and displays the formatted report.

"""
main.py
CLI entrypoint for the 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:
    """Format the report for terminal display."""
    lines = []
    lines.append("")
    lines.append("╔" + "═" * 58 + "╗")
    lines.append("║" + "  📄 RESEARCH REPORT".center(58) + "║")
    lines.append("╚" + "═" * 58 + "╝")

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

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

    lines.append(f"\n{'─' * 60}")
    lines.append("🔍 SUB-QUERIES RESEARCHED")
    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("💡 MAIN FINDINGS")
    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%} confidence]")
        lines.append(f"     {finding['description']}")

    lines.append(f"\n{'─' * 60}")
    lines.append(f"📚 SOURCES CONSULTED ({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():
    """Interactive mode: the user types topics."""
    print("=" * 60)
    print("  🔬 AI Research Assistant v1")
    print("  Type a topic to research.")
    print("  Commands: 'exit' to quit")
    print("=" * 60)

    while True:
        try:
            topic = input("\n🔎 Topic: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nSee you later!")
            break

        if not topic:
            continue

        if topic.lower() in ("exit", "quit", "q"):
            print("\nSee you later!")
            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 during the research: {e}")
            print("   Try another topic.")


def run_single(topic: str):
    """Run a single research pass."""
    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📦 Report 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()

The CLI has two modes:

Interactive: python main.py — a loop where the user types topics
Single shot: python main.py "impact of LLMs on education" — one research pass and it exits

The report is displayed formatted for the terminal and also as raw JSON. The JSON is useful for verifying that the structured output is valid.

Step 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

Complete code, file by file

For quick reference, here are all the files consolidated. If you followed the steps above, you already have them. If you'd rather copy and run directly, here they are.

state/research_state.py

"""
state/research_state.py
Data models for the AI Research Assistant.
"""

from pydantic import BaseModel, Field
from datetime import datetime


class Source(BaseModel):
    name: str = Field(description="Name of the source")
    source_type: str = Field(description="Type: web, academic, news")
    content: str = Field(description="Content extracted from the source")


class KeyFinding(BaseModel):
    title: str = Field(description="Title of the finding")
    description: str = Field(description="Description of the finding")
    confidence: float = Field(
        description="Confidence in the finding (0.0 to 1.0)",
        ge=0.0,
        le=1.0,
    )


class ResearchReport(BaseModel):
    topic: str = Field(description="Topic researched")
    summary: str = Field(description="Executive summary (2-3 sentences)")
    key_findings: list[KeyFinding] = Field(
        description="Main findings",
        min_length=1,
    )
    sources: list[Source] = Field(
        description="Sources consulted",
        min_length=1,
    )
    sub_queries: list[str] = Field(
        description="Sub-questions generated for the research",
    )
    confidence: float = Field(
        description="Overall confidence of the report (0.0 to 1.0)",
        ge=0.0,
        le=1.0,
    )
    generated_at: str = Field(
        default_factory=lambda: datetime.now().isoformat(),
        description="Generation timestamp",
    )


class SubQuery(BaseModel):
    query: str = Field(description="The sub-question")
    rationale: str = Field(description="Why this question is relevant")

config/settings.py

"""
config/settings.py
Configuration for the 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 for the AI Research Assistant.
"""

import hashlib
from langgraph.func import task


MOCK_RESULTS = {
    "web": {
        "default": (
            "Multiple web sources agree that {query} is a topic of growing "
            "interest. Experts highlight significant advances over the last 2 years. "
            "Practical applications include automation, data analysis "
            "and content generation."
        ),
    },
    "academic": {
        "default": (
            "Recent research (2025-2026) shows that {query} has solid theoretical "
            "foundations backed by multiple peer-reviewed studies. "
            "Experimental results demonstrate 40-60% improvements on key "
            "metrics compared to traditional methods."
        ),
    },
    "news": {
        "default": (
            "Recent news reports that {query} is making an impact across the "
            "industry. Leading companies like Google, Microsoft and emerging startups "
            "are investing heavily in this area. Major developments are expected "
            "by the end of 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:
    """Search the general web (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:
    """Search academic sources (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:
    """Search recent news (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 for the 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
Main agent of the 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]:
    """Break a research topic down into specific sub-questions."""
    response = model.invoke(
        f"You are an expert researcher. Break this topic down into "
        f"{MAX_SUB_QUERIES} specific, researchable sub-questions.\n\n"
        f"Topic: {topic}\n\n"
        f"Answer in JSON (no markdown, no ```json):\n"
        f'[{{"query": "sub-question", "rationale": "why it is relevant"}}]\n\n'
        f"Only the JSON, nothing else."
    )
    try:
        queries = json.loads(response.content)
        return queries[:MAX_SUB_QUERIES]
    except json.JSONDecodeError:
        return [
            {"query": topic, "rationale": "Original query as a fallback"},
            {"query": f"recent advances in {topic}", "rationale": "Current trends"},
            {"query": f"practical applications of {topic}", "rationale": "Real-world use"},
        ]


@task
def search_all_sources(query: str) -> list[dict]:
    """Search every configured source for one 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]:
    """Synthesize search results into key findings."""
    results_text = ""
    for i, result in enumerate(all_results, 1):
        results_text += (
            f"\nSource {i} ({result['source_type']}): {result['content']}\n"
        )

    response = model.invoke(
        f"You are a research analyst. Based on these sources, "
        f"identify 3-5 key findings about '{topic}'.\n\n"
        f"Sources:\n{results_text}\n\n"
        f"Answer in JSON (no markdown, no ```json):\n"
        f'[{{"title": "short title", "description": "1-2 sentence description", '
        f'"confidence": 0.8}}]\n\n'
        f"confidence goes from 0.0 to 1.0. Only the JSON, nothing else."
    )
    try:
        findings = json.loads(response.content)
        return findings[:5]
    except json.JSONDecodeError:
        return [{
            "title": "General finding",
            "description": f"Research on {topic} shows relevant results.",
            "confidence": 0.6,
        }]


@task
def generate_summary(topic: str, findings: list[dict]) -> str:
    """Generate a 2-3 sentence executive summary."""
    findings_text = "\n".join(
        f"- {f['title']}: {f['description']}" for f in findings
    )
    response = model.invoke(
        f"Generate a 2-3 sentence executive summary of the research "
        f"on the topic '{topic}'.\n\n"
        f"Main findings:\n{findings_text}\n\n"
        f"Only the summary, no title and no extra formatting."
    )
    return response.content.strip()


memory = MemorySaver()


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

    # Step 1: Decompose
    print(f"\n📋 Step 1: Decomposing topic into 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 generated:")
    for i, sq in enumerate(sub_queries, 1):
        print(f"     {i}. {sq.query}")

    # Step 2: Search in parallel
    print(f"\n🔍 Step 2: Searching {len(SEARCH_SOURCES)} sources per 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)} sources consulted")

    print(f"   Total: {len(all_results)} results collected")

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

    # Step 4: Summary
    print(f"\n📝 Step 4: Generating executive summary...")
    summary = generate_summary(topic, findings_raw).result()
    print(f"   ✓ Summary generated ({len(summary)} chars)")

    # Step 5: Confidence
    print(f"\n📊 Step 5: Computing the report's confidence...")
    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"   ✓ Confidence: {confidence:.0%}")

    # Step 6: Report
    print(f"\n📄 Step 6: Building the report...")
    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"   ✓ Report generated successfully")
    print(f"\n{'=' * 60}")

    return report.model_dump()

main.py

"""
main.py
CLI entrypoint for the 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("║" + "  📄 RESEARCH REPORT".center(58) + "║")
    lines.append("╚" + "═" * 58 + "╝")

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

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

    lines.append(f"\n{'─' * 60}")
    lines.append("🔍 SUB-QUERIES RESEARCHED")
    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("💡 MAIN FINDINGS")
    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%} confidence]")
        lines.append(f"     {finding['description']}")

    lines.append(f"\n{'─' * 60}")
    lines.append(f"📚 SOURCES CONSULTED ({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("  Type a topic to research.")
    print("  Commands: 'exit' to quit")
    print("=" * 60)

    while True:
        try:
            topic = input("\n🔎 Topic: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n\nSee you later!")
            break

        if not topic:
            continue

        if topic.lower() in ("exit", "quit", "q"):
            print("\nSee you later!")
            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 during the research: {e}")
            print("   Try another topic.")


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📦 Report 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()

Running it

Single mode (one research pass)

cd research-assistant
python main.py "impact of artificial intelligence on education"

Interactive mode

cd research-assistant
python main.py

Success criteria

Your project is complete when you meet these criteria:

✅ The agent decomposes the topic into 3+ sub-queries — the LLM generates specific, relevant sub-questions from the given topic
✅ The searches run in parallel — the 3 sources (web, academic, news) are searched simultaneously for each sub-query using the futures pattern
✅ The final report is structured (Pydantic) — the output is a valid ResearchReport with every required field: topic, summary, key_findings, sources, sub_queries, confidence
✅ The streaming shows progress — during execution you see each step complete: decomposition, search, synthesis, summary, confidence, report
✅ The CLI works in both modes — interactive (no arguments) and single-shot (with an argument)
✅ The report's JSON is valid — you can copy the JSON output and parse it without errors

Test scenarios

Test 1: A technical topic

🔎 Topic: machine learning applied to medical diagnosis

============================================================
  🔬 AI Research Assistant v1
  Topic: machine learning applied to medical diagnosis
============================================================

📋 Step 1: Decomposing topic into sub-queries...
   ✓ 4 sub-queries generated:
     1. Which ML algorithms are most used in medical diagnosis?
     2. What is the current accuracy of ML models in disease detection?
     3. What ethical challenges does the use of ML in medicine raise?
     4. Which hospitals or institutions are deploying ML in diagnosis?

🔍 Step 2: Searching 3 sources per sub-query...
   ✓ Sub-query 1: 3 sources consulted
   ✓ Sub-query 2: 3 sources consulted
   ✓ Sub-query 3: 3 sources consulted
   ✓ Sub-query 4: 3 sources consulted
   Total: 12 results collected

🧠 Step 3: Synthesizing findings...
   ✓ 4 findings identified:
     1. [85%] Predominant algorithms in diagnosis
     2. [80%] Accuracy comparable to specialists
     3. [75%] Significant ethical challenges
     4. [70%] Growing institutional adoption

📝 Step 4: Generating executive summary...
   ✓ Summary generated (180 chars)

📊 Step 5: Computing the report's confidence...
   ✓ Confidence: 82%

📄 Step 6: Building the report...
   ✓ Report generated successfully

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

╔══════════════════════════════════════════════════════════╗
║                     📄 RESEARCH REPORT                    ║
╚══════════════════════════════════════════════════════════╝

📌 Topic: machine learning applied to medical diagnosis
📅 Generated: 2026-03-08T...
🎯 Confidence: 82%

────────────────────────────────────────────────────────────
📋 EXECUTIVE SUMMARY
────────────────────────────────────────────────────────────
Applying machine learning to medical diagnosis shows significant
advances, with accuracy comparable to specialists...

────────────────────────────────────────────────────────────
💡 MAIN FINDINGS
────────────────────────────────────────────────────────────

  1. Predominant algorithms in diagnosis [85% confidence]
     Deep learning and convolutional neural networks lead...

Test 2: A general topic

🔎 Topic: remote work trends 2026

[The agent generates sub-queries like: labor regulations,
productivity tools, impact on mental health,
hybrid models, etc. The report synthesizes web,
academic and news sources.]

Test 3: A short, vague topic

🔎 Topic: Python

[The agent breaks "Python" down into more specific sub-queries:
history and evolution, main applications, comparison
with other languages, library ecosystem. It shows that
the agent adds value even with vague input.]

Test 4: Verify the structured JSON

python main.py "renewable energy" 2>/dev/null | grep -A 999 "Report JSON" | python -m json.tool

If the JSON parses without errors, the structured output is valid.

Common errors

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

Cause: Python can't find the modules because you're not running from the right directory.

Fix: Always run from the project root:

cd research-assistant
python main.py

The sys.path.insert(0, ".") in the files makes sure Python looks for modules in the current directory. If you run from another directory, it won't find them.

2. `json.JSONDecodeError` when parsing the LLM's response

Cause: The LLM sometimes wraps the JSON in markdown (```json ... ```) or adds text before/after. This is especially common with smaller models.

Fix: The code already has fallbacks for this case. If it happens often, you can add a cleanup step:

import re

def clean_json(text: str) -> str:
    """Extract JSON from a response that may include markdown."""
    match = re.search(r'\[.*\]', text, re.DOTALL)
    if match:
        return match.group(0)
    return text

3. The agent generates only 1-2 sub-queries instead of 4

Cause: The LLM may generate fewer queries than requested if it considers the topic simple, or if the JSON parsing truncates the results.

Fix: The fallback in decompose_query generates 3 queries by default if the parsing fails. If you need exactly N queries, you can add validation:

while len(queries) < MAX_SUB_QUERIES:
    queries.append({
        "query": f"another aspect of {topic}",
        "rationale": "Fill the minimum number of sub-queries",
    })

4. Pydantic `ValidationError` when building the report

Cause: Some field of the report fails validation (e.g. confidence outside the 0-1 range, empty key_findings, etc.).

Fix: Check which field fails in the error message. The most common errors:

# confidence out of range
confidence = max(0.0, min(1.0, calculated_value))

# empty key_findings
if not findings:
    findings = [KeyFinding(
        title="No specific findings",
        description="No findings could be identified with the available information.",
        confidence=0.3,
    )]

5. The searches don't run in parallel

Cause: If the checkpointer isn't active or there's an error in how the futures are set up, the tasks run sequentially.

Fix: Check that you're using the right futures pattern — launch every task first, then collect the results:

# ✅ Correct: launch them all first
futures = [search_all_sources(sq.query) for sq in sub_queries]
results = [f.result() for f in futures]

# ❌ Wrong: immediate result (sequential)
results = [search_all_sources(sq.query).result() for sq in sub_queries]

6. `OPENAI_API_KEY` not found

Cause: The .env file doesn't exist, isn't in the right directory, or the variable has a different name.

Fix: Check:

# Check that .env exists in the project directory
ls -la research-assistant/.env

# Check the content (without revealing the full key)
head -c 20 research-assistant/.env

7. The report always has the same confidence

Cause: The calculate_confidence function uses fixed metrics (number of sources, average relevance, number of findings). With mock searches, those metrics barely vary.

Fix: That's expected in v1 with mocks. When you replace the mocks with real search in M7, the metrics will vary naturally with the actual quality of the results.

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

Cause: You're trying to access a field of the result before calling .result():

# ❌ Error: a future isn't a dict
result = search_web(query)
content = result["content"]

# ✅ Correct: call .result() first
result = search_web(query).result()
content = result["content"]

What's next: the Research Assistant's evolution

What you built today is v1. It works end-to-end, but it has obvious limitations. Each following module solves one of them:

Module	Current limitation	What gets added
M7: Advanced Flows	The mock searches never fail	Retry logic with backoff, robust error handling, real search with fallback
M8: Memory	Every research run starts from scratch	Persistence with PostgresSaver, summarizing long research runs, memory of user preferences
M9: Human-in-the-Loop	The agent acts without supervision	Approval before expensive actions, human review of the synthesis, editable state
M10: Multi-Agent	A single agent does everything	4 specialized agents (researcher, analyst, writer, supervisor) working together
M11: Deep Agents	Planning is hardcoded	Autonomous planning with `write_todos`, a filesystem to store research, subagent spawning
M12: Production	No observability	LangSmith tracing, evaluation datasets, token tracking, rate limiting

The v1 has the right architecture to support this evolution. The Pydantic models will get extended. The @tasks will be replaced with more robust versions. The @entrypoint will eventually become a hybrid system with StateGraph for the steps that need it. The file structure (agents/, tools/, state/, config/) already anticipates everything that's coming.

Don't delete what you built today. You'll be iterating on it for 6 more modules.

Project resources

LangGraph Functional API Guide — Official documentation for @entrypoint and @task
LangGraph Checkpointing — How MemorySaver and checkpointing work
Pydantic v2 Documentation — Data models, validation, serialization
LangChain init_chat_model — Model initialization
LangGraph How-To Guides — Practical patterns for workflows
Python asyncio and Futures — Futures concepts in Python (analogous to LangGraph's futures)

Module 6 — LangChain & LangGraph: From Chains to Agents

← PreviousCombining the Graph and Functional APIs Next →Introduction: Real-World Complex Workflows

Evolving Project: Research Agent Baseline (v1)

Project overview

Project goal

Technical specs

Tech stack

Initial setup

Project structure

Step 1: Data models (state/research_state.py)

Step 2: Configuration (config/settings.py)

Step 3: Tools — search and calculation (tools/)

tools/web_search.py

tools/calculator.py

Step 4: The main agent (agents/researcher.py)

Step 5: CLI (main.py)

Step 6: requirements.txt

Complete code, file by file

state/research_state.py

config/settings.py

tools/web_search.py

tools/calculator.py

agents/researcher.py

main.py

Running it

Single mode (one research pass)

Interactive mode

Success criteria

Test scenarios

Test 1: A technical topic

Test 2: A general topic

Test 3: A short, vague topic

Test 4: Verify the structured JSON

Common errors

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

2. json.JSONDecodeError when parsing the LLM's response

3. The agent generates only 1-2 sub-queries instead of 4

4. Pydantic ValidationError when building the report

5. The searches don't run in parallel

6. OPENAI_API_KEY not found

7. The report always has the same confidence

8. TypeError: 'Future' object is not subscriptable

What's next: the Research Assistant's evolution

Project resources

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

2. `json.JSONDecodeError` when parsing the LLM's response

4. Pydantic `ValidationError` when building the report

6. `OPENAI_API_KEY` not found

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