Combining the Graph and Functional APIs

Capsule overview

You don't have to marry a single API. LangGraph was designed so that the Graph API and the Functional API coexist in the same system — and that's one of its most powerful design decisions. You can use a @task inside a StateGraph node, call a compiled graph from an @entrypoint, or compose @entrypoints that call each other. Each part of your system uses the tool that best expresses its logic.

Why would you want to mix them? Because real systems aren't uniform. Your main flow may be sequential (perfect for @entrypoint), but one of the steps may have complex conditional routing (perfect for StateGraph). Or your graph may have a node that internally needs to run checkpointable sub-operations (perfect for @task). Combining them isn't a hack — it's the intended use case.

In this capsule you're going to learn three combination patterns, see a real system that uses them together, and build the judgment to know when combining makes sense and when it's over-engineering. This is the last step before the evolving project — the AI Research Assistant you'll build in the next capsule will use exactly this kind of hybrid architecture as it grows.

Pattern 1: @task inside a StateGraph

The first pattern is the most granular one: you use @task inside a function that serves as a node of a StateGraph. The graph handles routing and the macro structure, while @task gives you checkpointable sub-operations inside an individual node.

When to use this pattern

• ✅ Your workflow has conditional routing (it needs the Graph API)
• ✅ An individual node needs to run sub-steps that you want to be checkpointable
• ✅ You want the sub-operations inside a node to survive crashes

The concept

Normally, a StateGraph node is just a regular function. Everything that happens inside the node is one monolithic operation — if it fails halfway through, it repeats from the beginning. But if you use @task inside the node, each sub-operation gets recorded. If the node fails after completing 2 of 3 tasks, only the third one is re-executed when you resume.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

model = init_chat_model("openai:gpt-4.1-mini")


@task
def extract_keywords(text: str) -> list[str]:
    """Extract keywords from a text using the LLM."""
    response = model.invoke(
        f"Extract 3-5 keywords from the following text. "
        f"Return only the keywords, separated by commas.\n\n{text}"
    )
    return [kw.strip() for kw in response.content.split(",")]


@task
def score_relevance(text: str, topic: str) -> float:
    """Compute a relevance score from 0 to 1."""
    response = model.invoke(
        f"On a scale from 0.0 to 1.0, how relevant is this text "
        f"to the topic '{topic}'? Answer with the number ONLY.\n\n{text}"
    )
    try:
        return float(response.content.strip())
    except ValueError:
        return 0.5


class AnalysisState(TypedDict):
    text: str
    topic: str
    keywords: list[str]
    relevance_score: float
    analysis: str


def analyze_node(state: AnalysisState) -> dict:
    """StateGraph node that internally uses @tasks."""
    keywords_future = extract_keywords(state["text"])
    score_future = score_relevance(state["text"], state["topic"])

    keywords = keywords_future.result()
    score = score_future.result()

    return {
        "keywords": keywords,
        "relevance_score": score,
    }


def summarize_node(state: AnalysisState) -> dict:
    """Generate a summary based on the analysis."""
    kw_str = ", ".join(state["keywords"])
    response = model.invoke(
        f"Summarize this text in 2 sentences. "
        f"Keywords identified: {kw_str}. "
        f"Relevance to the topic '{state['topic']}': {state['relevance_score']}\n\n"
        f"{state['text']}"
    )
    return {"analysis": response.content}


def route_by_relevance(state: AnalysisState) -> str:
    """If relevance is high, generate a summary. Otherwise, finish."""
    if state["relevance_score"] >= 0.5:
        return "summarize"
    return "__end__"


graph = StateGraph(AnalysisState)
graph.add_node("analyze", analyze_node)
graph.add_node("summarize", summarize_node)

graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", route_by_relevance)
graph.add_edge("summarize", END)

app = graph.compile()

result = app.invoke({
    "text": "Python is a high-level, interpreted, general-purpose programming language. It is widely used in artificial intelligence, data science and web development.",
    "topic": "artificial intelligence",
})

print(f"Keywords: {result['keywords']}")
print(f"Relevance: {result['relevance_score']}")
print(f"Analysis: {result['analysis']}")
# Expected output:
# Keywords: ['Python', 'artificial intelligence', 'data science', 'programming', 'web development']
# Relevance: 0.8
# Analysis: Python is a general-purpose language widely used in AI and data science...

The key point: analyze_node is an ordinary graph node, but internally it launches extract_keywords and score_relevance as @tasks. If you add a checkpointer to the graph, those tasks get recorded individually.

What does NOT work directly

An important detail: @task is designed to run inside an @entrypoint context. If your StateGraph doesn't have an @entrypoint as a wrapper, the tasks still run, but without the full benefit of individual checkpointing. To get real checkpointing of the sub-tasks, you need Pattern 2, or you wrap the node in an @entrypoint.

The most practical way to turn this on is to compile the graph with a checkpointer:

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

result = app.invoke(
    {
        "text": "LangGraph lets you build agents as state graphs.",
        "topic": "AI agents",
    },
    config={"configurable": {"thread_id": "analysis-1"}},
)
print(f"Keywords: {result['keywords']}")
print(f"Analysis: {result['analysis']}")
# Expected output:
# Keywords: ['LangGraph', 'agents', 'graphs', 'state', 'AI']
# Analysis: LangGraph is a framework for building agents based on state graphs...

Pattern 2: A compiled graph inside an @entrypoint

The second pattern is the inverse: your main flow is an @entrypoint (sequential, easy to read), but one of the steps is complex enough to justify a StateGraph. So you call the compiled graph as if it were any other function.

When to use this pattern

• ✅ Your main flow is sequential (plan → execute → analyze → report)
• ✅ One of the steps has conditional routing or internal loops
• ✅ You want to encapsulate a step's complexity in a graph without letting it "contaminate" the main flow

Example: a sequential flow with one complex step

Imagine a system that: 1) receives a question, 2) searches across multiple sources (a complex step with routing), 3) synthesizes the answer. Step 2 is a graph; steps 1 and 3 are simple functions.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

model = init_chat_model("openai:gpt-4.1-mini")


# --- Complex step: multi-source search graph ---

class SearchState(TypedDict):
    query: str
    source_type: str
    results: Annotated[list[str], operator.add]


def classify_source(state: SearchState) -> dict:
    query_lower = state["query"].lower()
    if any(w in query_lower for w in ["code", "api", "function", "python"]):
        return {"source_type": "docs"}
    elif any(w in query_lower for w in ["latest", "recent", "news", "2026"]):
        return {"source_type": "news"}
    return {"source_type": "general"}


def search_docs(state: SearchState) -> dict:
    return {"results": [f"[DOCS] Technical documentation about: {state['query']}"]}


def search_news(state: SearchState) -> dict:
    return {"results": [f"[NEWS] Latest news about: {state['query']}"]}


def search_general(state: SearchState) -> dict:
    return {"results": [f"[WEB] General information about: {state['query']}"]}


def route_source(state: SearchState) -> str:
    return {
        "docs": "search_docs",
        "news": "search_news",
        "general": "search_general",
    }[state["source_type"]]


search_graph = StateGraph(SearchState)
search_graph.add_node("classify", classify_source)
search_graph.add_node("search_docs", search_docs)
search_graph.add_node("search_news", search_news)
search_graph.add_node("search_general", search_general)

search_graph.add_edge(START, "classify")
search_graph.add_conditional_edges("classify", route_source)
search_graph.add_edge("search_docs", END)
search_graph.add_edge("search_news", END)
search_graph.add_edge("search_general", END)

compiled_search = search_graph.compile()


# --- Main flow: sequential @entrypoint ---

@task
def plan_search(question: str) -> str:
    """Plan what to search for based on the question."""
    response = model.invoke(
        f"Given this question, generate an optimized search query "
        f"(1 line, no explanation):\n\n{question}"
    )
    return response.content.strip()


@task
def execute_search(query: str) -> list[str]:
    """Run the search using the compiled graph."""
    result = compiled_search.invoke({"query": query})
    return result["results"]


@task
def synthesize(question: str, search_results: list[str]) -> str:
    """Synthesize the results into an answer."""
    results_text = "\n".join(search_results)
    response = model.invoke(
        f"Original question: {question}\n\n"
        f"Search results:\n{results_text}\n\n"
        f"Generate a clear and complete answer."
    )
    return response.content


memory = MemorySaver()


@entrypoint(checkpointer=memory)
def research_flow(question: str) -> str:
    """Main research flow."""
    optimized_query = plan_search(question).result()
    print(f"  Optimized query: {optimized_query}")

    results = execute_search(optimized_query).result()
    print(f"  Results found: {len(results)}")

    answer = synthesize(question, results).result()
    return answer


response = research_flow.invoke(
    "What's the latest news in Python 3.13?",
    config={"configurable": {"thread_id": "research-1"}},
)
print(f"\nAnswer:\n{response}")
# Expected output:
#   Optimized query: Python 3.13 new features latest news
#   Results found: 1
#
# Answer:
# The latest news in Python 3.13 includes improvements in...

The @entrypoint reads like plain Python: plan → search → synthesize. But the search step internally uses a StateGraph with conditional routing. The main flow doesn't know and doesn't care that compiled_search is a graph — it treats it like any other function.

The advantage of this pattern

The readability of the main flow stays clean. If you read research_flow, you see three sequential steps. You don't need to understand the search graph to understand the overall flow. The complexity is encapsulated.

Pattern 3: An @entrypoint calling another @entrypoint

The third pattern is modular composition: an @entrypoint that calls another @entrypoint. Each module in your system is an independent workflow with its own checkpointing, and the main workflow orchestrates them.

When to use this pattern

• ✅ Your system has independent modules that could run on their own
• ✅ Each module needs its own checkpointing and error handling
• ✅ You want to reuse modules across different flows
• ✅ Different teams work on different modules

Example: a modular content pipeline

from dotenv import load_dotenv
load_dotenv()

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

model = init_chat_model("openai:gpt-4.1-mini")


# --- Module 1: Research ---

@task
def gather_info(topic: str) -> str:
    response = model.invoke(
        f"Briefly research: {topic}. "
        f"Give 3 key points in list format."
    )
    return response.content


@entrypoint()
def research_module(topic: str) -> dict:
    """Independent research module."""
    info = gather_info(topic).result()
    return {"topic": topic, "findings": info}


# --- Module 2: Analysis ---

@task
def analyze_findings(findings: str) -> str:
    response = model.invoke(
        f"Analyze these findings and identify the main trend:\n\n"
        f"{findings}"
    )
    return response.content


@entrypoint()
def analysis_module(research_data: dict) -> dict:
    """Independent analysis module."""
    analysis = analyze_findings(research_data["findings"]).result()
    return {
        "topic": research_data["topic"],
        "findings": research_data["findings"],
        "analysis": analysis,
    }


# --- Module 3: Writing ---

@task
def write_report(topic: str, findings: str, analysis: str) -> str:
    response = model.invoke(
        f"Write a mini-report (3 paragraphs) about '{topic}'.\n\n"
        f"Findings:\n{findings}\n\n"
        f"Analysis:\n{analysis}"
    )
    return response.content


@entrypoint()
def writing_module(analysis_data: dict) -> str:
    """Independent writing module."""
    report = write_report(
        analysis_data["topic"],
        analysis_data["findings"],
        analysis_data["analysis"],
    ).result()
    return report


# --- Main orchestrator ---

@entrypoint()
def content_pipeline(topic: str) -> str:
    """Full pipeline: research → analyze → write."""
    research_data = research_module.invoke(topic)
    print(f"  Research complete: {len(research_data['findings'])} chars")

    analysis_data = analysis_module.invoke(research_data)
    print(f"  Analysis complete: {len(analysis_data['analysis'])} chars")

    report = writing_module.invoke(analysis_data)
    print(f"  Report generated: {len(report)} chars")

    return report


result = content_pipeline.invoke("the impact of generative AI on education")
print(f"\n{'=' * 60}")
print(result)
# Expected output:
#   Research complete: ~200 chars
#   Analysis complete: ~150 chars
#   Report generated: ~400 chars
#
# ============================================================
# [3-paragraph mini-report about generative AI in education]

Each module (research_module, analysis_module, writing_module) is a complete @entrypoint that could run independently. content_pipeline composes them by calling .invoke() on each one.

The difference from plain functions

Why not use plain functions instead of an @entrypoint for each module? Because @entrypoint gives you:

• ✅ Individual checkpointing per module
• ✅ Streaming of each module's progress
• ✅ Automatic resume if one of the modules fails
• ✅ A standard interface (.invoke(), .stream()) for each module

If a module is simple (a single model call), use @task. If it's a workflow with multiple steps, use @entrypoint.

A real architecture: hybrid research system

Now let's look at a more realistic example that combines all three patterns. A research system where:

• Main flow: a sequential @entrypoint (plan → research → analyze → write)
• Research step: a StateGraph with routing between sources
• Individual operations: a @task for each search (checkpointable, parallel)

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

model = init_chat_model("openai:gpt-4.1-mini")


# --- Individual @tasks for searches ---

@task
def search_academic(query: str) -> str:
    """Search academic sources (mock)."""
    response = model.invoke(
        f"Simulate an academic search result about: {query}. "
        f"Include a paper title and a key finding."
    )
    return f"[ACADEMIC] {response.content}"


@task
def search_web(query: str) -> str:
    """Search the general web (mock)."""
    return f"[WEB] Web results for '{query}': general information available across multiple sources."


@task
def search_news(query: str) -> str:
    """Search recent news (mock)."""
    return f"[NEWS] Latest news about '{query}': recent developments reported by specialized outlets."


# --- StateGraph for the research step with routing ---

class ResearchStepState(TypedDict):
    query: str
    depth: str
    results: Annotated[list[str], operator.add]


def assess_depth(state: ResearchStepState) -> dict:
    """Determine how deep the search needs to be."""
    query_lower = state["query"].lower()
    if any(w in query_lower for w in ["detailed", "deep", "complete", "exhaustive"]):
        return {"depth": "deep"}
    return {"depth": "standard"}


def standard_search(state: ResearchStepState) -> dict:
    """Standard search: web only."""
    web_result = search_web(state["query"]).result()
    return {"results": [web_result]}


def deep_search(state: ResearchStepState) -> dict:
    """Deep search: web + academic + news in parallel."""
    web_future = search_web(state["query"])
    academic_future = search_academic(state["query"])
    news_future = search_news(state["query"])

    return {
        "results": [
            web_future.result(),
            academic_future.result(),
            news_future.result(),
        ]
    }


def route_depth(state: ResearchStepState) -> str:
    return "deep_search" if state["depth"] == "deep" else "standard_search"


research_step = StateGraph(ResearchStepState)
research_step.add_node("assess", assess_depth)
research_step.add_node("standard_search", standard_search)
research_step.add_node("deep_search", deep_search)

research_step.add_edge(START, "assess")
research_step.add_conditional_edges("assess", route_depth)
research_step.add_edge("standard_search", END)
research_step.add_edge("deep_search", END)

compiled_research = research_step.compile()


# --- Main @entrypoint ---

@task
def plan_research(topic: str) -> list[str]:
    """Break the topic down into sub-queries."""
    response = model.invoke(
        f"Break this research topic into 2-3 specific sub-questions. "
        f"Return only the questions, one per line.\n\nTopic: {topic}"
    )
    queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return queries[:3]


@task
def synthesize_report(topic: str, all_results: list[str]) -> str:
    """Synthesize all results into a report."""
    results_text = "\n".join(f"- {r}" for r in all_results)
    response = model.invoke(
        f"Generate a research report about '{topic}'.\n\n"
        f"Sources collected:\n{results_text}\n\n"
        f"The report must have: an executive summary (2 sentences), "
        f"3 main findings, and a conclusion."
    )
    return response.content


memory = MemorySaver()


@entrypoint(checkpointer=memory)
def hybrid_research(topic: str) -> str:
    """Hybrid research system: @entrypoint + StateGraph + @tasks."""
    print(f"  Researching: {topic}")

    queries = plan_research(topic).result()
    print(f"  Sub-queries generated: {len(queries)}")

    all_results = []
    for query in queries:
        search_result = compiled_research.invoke({"query": query})
        all_results.extend(search_result["results"])
        print(f"    ✓ '{query[:40]}...' → {len(search_result['results'])} results")

    report = synthesize_report(topic, all_results).result()
    print(f"  Report generated: {len(report)} characters")

    return report


result = hybrid_research.invoke(
    "detailed analysis of the impact of LLMs on software development",
    config={"configurable": {"thread_id": "hybrid-1"}},
)
print(f"\n{'=' * 60}")
print(result)
# Expected output:
#   Researching: detailed analysis of the impact of LLMs on software development
#   Sub-queries generated: 3
#     ✓ 'How have LLMs changed day-to-day engin...' → 3 results
#     ✓ 'Which LLM-based coding tools are most...' → 3 results
#     ✓ 'What are the risks and limitations of...' → 3 results
#   Report generated: ~500 characters
#
# ============================================================
# [Report with executive summary, findings and conclusion]

Notice how the keyword "detailed" in the topic triggers the deep search (3 sources in parallel) instead of the standard one (web only). The @entrypoint controls the macro flow, the StateGraph handles the search routing, and the @tasks run the individual searches.

When combining makes sense vs when it's over-engineering

Do combine when:

• ✅ Your main flow is sequential but one step has complex routing
• ✅ Different parts of the system have different checkpointing needs
• ✅ You want teams working on independent modules
• ✅ An existing module (like an already battle-tested StateGraph) needs to be integrated into a new flow
• ✅ You need to mix parallel execution (@task futures) with conditional routing (StateGraph)

Don't combine when:

• ❌ Your whole flow is sequential → use only @entrypoint + @task
• ❌ Your whole flow has complex routing → use only StateGraph
• ❌ You're combining "just in case" or "to be prepared" → YAGNI
• ❌ Every "module" is a single LLM call → it doesn't need to be its own @entrypoint
• ❌ Your team is one person and the system has 3 nodes → modularity buys you nothing

The practical rule

Ask yourself: "does this step have a different complexity from the main flow?" If the answer is yes, encapsulate it with the API that expresses it best. If everything has the same complexity, use a single API.

Comparison: Pure Graph vs Pure Functional vs Hybrid

Recommended progression

1. Start with @entrypoint + @task (Functional API)
   → Most workflows start out sequential

2. If a step needs conditional routing, extract it into a StateGraph
   → The graph encapsulates that complexity

3. If you need independent modules, use @entrypoint as wrappers
   → Each module gets its own lifecycle

4. Result: a hybrid system that uses each tool where it shines

Preview: the AI Research Assistant and hybrid architecture

The project you'll build in the next capsule (and that will keep evolving until Module 12) starts simple with the Functional API. But as it grows, it will naturally need to combine:

The Research Assistant will evolve from pure Functional (M6) to a hybrid system (M7+). That's not a design accident — it's the natural progression. Start simple, add complexity only when the problem demands it.

Troubleshooting

1. @task inside a StateGraph isn't checkpointed individually

Cause: @tasks need an @entrypoint context or an active checkpointer to record their results individually. If your StateGraph is compiled without a checkpointer, the tasks run, but as plain functions.

Fix: Compile the graph with a checkpointer and pass a thread_id:

memory = MemorySaver()
app = graph.compile(checkpointer=memory)
result = app.invoke(input_data, config={"configurable": {"thread_id": "my-thread"}})

2. Serialization error when passing data between @entrypoint and StateGraph

Cause: The StateGraph expects a dictionary with specific keys (your TypedDict), but you're passing it a different type (a string, a list, etc.).

Fix: Make sure what you pass to compiled_graph.invoke() matches the TypedDict schema exactly:

# ❌ Error: passing a string where a dict is expected
result = compiled_search.invoke("my query")

# ✅ Correct: passing a dict with the TypedDict's keys
result = compiled_search.invoke({"query": "my query"})

3. The @entrypoint can't reach the StateGraph's internal state

Cause: The compiled StateGraph returns its final state as a dictionary. The @entrypoint receives that dictionary, not the typed state.

Fix: Access the graph's results by dictionary key:

@entrypoint()
def my_flow(topic: str) -> str:
    graph_result = compiled_graph.invoke({"query": topic})
    results = graph_result["results"]  # ← access by key
    return synthesize(results).result()

4. Deadlock when combining @entrypoint with a StateGraph on a shared checkpointer

Cause: If the @entrypoint and the compiled StateGraph use the same checkpointer with the same thread_id, they can compete for locks.

Fix: Use different thread_ids for each level, or let the inner graph compile without a checkpointer of its own:

compiled_search = search_graph.compile()  # no checkpointer of its own

@entrypoint(checkpointer=memory)
def main_flow(topic: str) -> str:
    result = compiled_search.invoke({"query": topic})  # no thread_id
    return result["results"]

5. I don't know if I need hybrid or if I'm over-complicating things

Cause: No clear criterion. The temptation to combine "because I can" is strong.

Fix: Start with pure Functional API. If at some point you need conditional routing that an if/else doesn't express well, extract that step into a StateGraph. If you never reach that point, you don't need hybrid. Let the pain guide you, not the anticipation.

Exercises

Exercise 1: Identify the right pattern (Basic)

For each scenario, decide whether you'd use: (A) Pure Functional, (B) Pure Graph, or (C) Hybrid. Justify your answer.

Scenario 1: A pipeline that translates a text into 3 languages in parallel and then concatenates the results.

Scenario 2: A support system that classifies tickets (billing/tech/general), routes them to a specialized handler, and inside the technical handler runs a 3-step checkpointable diagnostic.

Scenario 3: A chatbot that processes messages sequentially: detect language → translate if needed → answer → translate the answer back.

Exercise 2: Convert a StateGraph to hybrid (Basic)

You have this purely linear StateGraph. Convert it to an @entrypoint with @tasks, keeping the same behavior.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class PipeState(TypedDict):
    text: str
    cleaned: str
    summarized: str
    formatted: str

def clean(state):
    return {"cleaned": state["text"].strip().lower()}

def summarize(state):
    return {"summarized": f"Summary of: {state['cleaned'][:50]}"}

def format_output(state):
    return {"formatted": f"📄 {state['summarized']}"}

g = StateGraph(PipeState)
g.add_node("clean", clean)
g.add_node("summarize", summarize)
g.add_node("format", format_output)
g.add_edge(START, "clean")
g.add_edge("clean", "summarize")
g.add_edge("summarize", "format")
g.add_edge("format", END)
app = g.compile()

from langgraph.func import entrypoint, task


@task
def clean(text: str) -> str:
    return text.strip().lower()


@task
def summarize(cleaned: str) -> str:
    return f"Summary of: {cleaned[:50]}"


@task
def format_output(summarized: str) -> str:
    return f"📄 {summarized}"


@entrypoint()
def text_pipeline(text: str) -> str:
    cleaned = clean(text).result()
    summarized = summarize(cleaned).result()
    formatted = format_output(summarized).result()
    return formatted


result = text_pipeline.invoke("  This Is A SAMPLE Text To Process  ")
print(result)
# Expected output:
# 📄 Summary of: this is a sample text to process

Exercise 3: Integrate an existing graph into an @entrypoint (Intermediate)

You have this sentiment classification StateGraph that already works. Integrate it inside an @entrypoint that: 1) receives a list of texts, 2) classifies each one with the graph, 3) returns a summary of how many are positive, negative and neutral.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SentimentState(TypedDict):
    text: str
    sentiment: str

def classify_sentiment(state: SentimentState) -> dict:
    text_lower = state["text"].lower()
    positive_words = ["good", "excellent", "great", "amazing", "perfect", "love"]
    negative_words = ["bad", "terrible", "horrible", "awful", "hate", "error"]
    pos = sum(1 for w in positive_words if w in text_lower)
    neg = sum(1 for w in negative_words if w in text_lower)
    if pos > neg:
        return {"sentiment": "positive"}
    elif neg > pos:
        return {"sentiment": "negative"}
    return {"sentiment": "neutral"}

sentiment_graph = StateGraph(SentimentState)
sentiment_graph.add_node("classify", classify_sentiment)
sentiment_graph.add_edge(START, "classify")
sentiment_graph.add_edge("classify", END)
compiled_sentiment = sentiment_graph.compile()

from langgraph.func import entrypoint, task


@task
def analyze_text(text: str) -> str:
    """Classify a text using the sentiment graph."""
    result = compiled_sentiment.invoke({"text": text})
    return result["sentiment"]


@entrypoint()
def batch_sentiment(texts: list[str]) -> dict:
    """Classify a list of texts and summarize the results."""
    futures = [analyze_text(text) for text in texts]
    sentiments = [f.result() for f in futures]

    summary = {
        "positive": sentiments.count("positive"),
        "negative": sentiments.count("negative"),
        "neutral": sentiments.count("neutral"),
        "total": len(sentiments),
        "details": list(zip(texts, sentiments)),
    }
    return summary


texts = [
    "This product is excellent, I love it",
    "Terrible experience, everything went bad",
    "The package arrived yesterday afternoon",
    "Amazing service, great support",
    "Horrible error in the payments system",
]

result = batch_sentiment.invoke(texts)
print(f"Total: {result['total']}")
print(f"Positive: {result['positive']}")
print(f"Negative: {result['negative']}")
print(f"Neutral: {result['neutral']}")
for text, sent in result["details"]:
    print(f"  [{sent:>8}] {text[:50]}")
# Expected output:
# Total: 5
# Positive: 2
# Negative: 2
# Neutral: 1
#   [positive] This product is excellent, I love it
#   [negative] Terrible experience, everything went bad
#   [ neutral] The package arrived yesterday afternoon
#   [positive] Amazing service, great support
#   [negative] Horrible error in the payments system

Exercise 4: Design a hybrid architecture (Intermediate)

Without writing the full code, design the architecture for a content generation system that:

1. Receives a topic and a desired format (blog, tweet thread, email newsletter)
2. Researches the topic (search across 2-3 sources)
3. Generates a draft in the right format
4. Validates the draft (appropriate length, correct tone)
5. If validation fails, regenerates it (maximum 2 attempts)

Define: which parts would use @entrypoint, which @task, which StateGraph, and why.

Proposed architecture:

1. MAIN FLOW: @entrypoint (sequential)
   → content_generator(topic, format)
   Reason: the macro flow is sequential (research → generate → validate)

2. RESEARCH STEP: @task (parallel)
   → search_source_1(topic), search_source_2(topic), search_source_3(topic)
   Reason: parallel searches with no conditional routing. Futures pattern.

3. GENERATION STEP: @task
   → generate_draft(topic, format, research_results)
   Reason: a single operation. It doesn't need to be its own module.

4. VALIDATION + RETRY STEP: StateGraph
   → validate_node → route(pass/fail) → regenerate_node → validate_node (loop)
   Reason: it has a conditional loop (validate → ok? → if not, regenerate → validate).
   The retry with a maximum of 2 attempts is modeled as a loop with a counter in the state.
   Conditional edge: if validation_pass=True → END, if attempts < 2 → regenerate.

   StateGraph(DraftState):
     - validate: checks length and tone
     - regenerate: generates a new draft with feedback
     - route: pass → END, fail+attempts<2 → regenerate, fail+attempts>=2 → END

5. COMPOSITION:
   @entrypoint content_generator:
     research = [search_1(topic), search_2(topic)]  # parallel @tasks
     draft = generate_draft(topic, format, results)  # @task
     final = compiled_validation_graph.invoke(draft)  # StateGraph with a loop

Why hybrid:
- The main flow is sequential → @entrypoint
- The searches are parallel with no routing → @task with futures
- The validation has a conditional loop → StateGraph
- Pure Functional doesn't handle the retry loop well
- Pure Graph would make the main flow needlessly verbose

Exercise 5: Implement nested @entrypoints with error handling (Advanced)

Implement two @entrypoint modules and one orchestrating @entrypoint:

• Module A (translator): Receives a text and a target language. Returns the translation (mock: prepends a prefix with the language).
• Module B (quality_checker): Receives a text and returns {"passed": True/False, "reason": "..."}. It fails (returns passed: False) if the text has fewer than 10 characters.
• Orchestrator: Translates the text, checks quality. If quality fails, it tries translating again with an improved prompt. Maximum 2 attempts.

from langgraph.func import entrypoint, task


@task
def do_translation(text: str, target_lang: str, attempt: int) -> str:
    """Translation mock. On attempt 2 it generates a longer text."""
    prefix = f"[{target_lang.upper()}]"
    if attempt > 1:
        return f"{prefix} (improved) Full and detailed translation of: {text}"
    return f"{prefix} {text}"


@entrypoint()
def translator(params: dict) -> str:
    """Translation module."""
    result = do_translation(
        params["text"],
        params["target_lang"],
        params.get("attempt", 1),
    ).result()
    return result


@task
def check_quality(text: str) -> dict:
    """Check quality: minimum 10 characters."""
    if len(text) < 10:
        return {"passed": False, "reason": f"Too short ({len(text)} chars, minimum 10)"}
    return {"passed": True, "reason": "OK"}


@entrypoint()
def quality_checker(text: str) -> dict:
    """Quality-check module."""
    return check_quality(text).result()


@entrypoint()
def translate_with_quality(params: dict) -> dict:
    """Orchestrator: translate and check quality, with retry."""
    text = params["text"]
    target_lang = params["target_lang"]
    max_attempts = 2

    for attempt in range(1, max_attempts + 1):
        translation = translator.invoke({
            "text": text,
            "target_lang": target_lang,
            "attempt": attempt,
        })
        print(f"  Attempt {attempt}: '{translation}'")

        quality = quality_checker.invoke(translation)
        print(f"  Quality: {quality}")

        if quality["passed"]:
            return {
                "translation": translation,
                "attempts": attempt,
                "quality": quality,
            }

        print(f"  Retry: {quality['reason']}")

    return {
        "translation": translation,
        "attempts": max_attempts,
        "quality": quality,
        "warning": "Quality not reached after the maximum number of attempts",
    }


result = translate_with_quality.invoke({
    "text": "Hi",
    "target_lang": "es",
})
print(f"\nResult: {result}")
# Expected output:
#   Attempt 1: '[ES] Hi'
#   Quality: {'passed': False, 'reason': 'Too short (7 chars, minimum 10)'}
#   Retry: Too short (7 chars, minimum 10)
#   Attempt 2: '[ES] (improved) Full and detailed translation of: Hi'
#   Quality: {'passed': True, 'reason': 'OK'}
#
# Result: {'translation': '[ES] (improved) ...', 'attempts': 2, 'quality': {'passed': True, ...}}

result2 = translate_with_quality.invoke({
    "text": "This is a long text that should pass",
    "target_lang": "fr",
})
print(f"\nResult 2: {result2}")
# Expected output:
#   Attempt 1: '[FR] This is a long text...'
#   Quality: {'passed': True, 'reason': 'OK'}
#
# Result 2: {'translation': '[FR] This is a long...', 'attempts': 1, ...}

Exercise 6: Refactor from hybrid to pure Functional (Challenge)

You have this hybrid system. The StateGraph routes only 2 paths and has no loops. Refactor all of it to pure Functional API (only @entrypoint + @task), replacing the conditional edge with an if/else. Compare the readability of the two versions.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task

class ProcessState(TypedDict):
    data: str
    category: str
    result: str

def categorize(state: ProcessState) -> dict:
    if len(state["data"]) > 20:
        return {"category": "complex"}
    return {"category": "simple"}

def handle_simple(state: ProcessState) -> dict:
    return {"result": f"Simple: {state['data'].upper()}"}

def handle_complex(state: ProcessState) -> dict:
    return {"result": f"Complex: deep analysis of '{state['data'][:20]}...'"}

def route_category(state: ProcessState) -> str:
    return "handle_complex" if state["category"] == "complex" else "handle_simple"

g = StateGraph(ProcessState)
g.add_node("categorize", categorize)
g.add_node("handle_simple", handle_simple)
g.add_node("handle_complex", handle_complex)
g.add_edge(START, "categorize")
g.add_conditional_edges("categorize", route_category)
g.add_edge("handle_simple", END)
g.add_edge("handle_complex", END)
process_graph = g.compile()

from langgraph.func import entrypoint, task


@task
def categorize(data: str) -> str:
    if len(data) > 20:
        return "complex"
    return "simple"


@task
def handle_simple(data: str) -> str:
    return f"Simple: {data.upper()}"


@task
def handle_complex(data: str) -> str:
    return f"Complex: deep analysis of '{data[:20]}...'"


@entrypoint()
def process_data(data: str) -> str:
    category = categorize(data).result()

    if category == "complex":
        result = handle_complex(data).result()
    else:
        result = handle_simple(data).result()

    return result


print(process_data.invoke("Hi"))
# Expected output: Simple: HI

print(process_data.invoke("This is a text long enough to count as complex"))
# Expected output: Complex: deep analysis of 'This is a text long ...'

Summary

In this capsule you learned:

• Pattern 1: @task inside a StateGraph — use the graph for macro routing and @task for checkpointable sub-operations inside a node
• Pattern 2: A compiled graph inside an @entrypoint — a sequential main flow with one complex step encapsulated as a graph. The complexity doesn't pollute the flow's readability
• Pattern 3: @entrypoint calling @entrypoint — modular composition where each module is an independent workflow with its own lifecycle
• Combining isn't a hack — it's an intended use case. LangGraph was designed for both APIs to coexist
• When to combine: when different parts of the system have different needs (routing vs sequential, granular checkpointing, independent teams)
• When NOT to combine: when everything is sequential, when everything is routing, or when you're anticipating future needs that don't exist today
• The AI Research Assistant will evolve from pure Functional (M6) to hybrid (M7+) — exactly as recommended: start simple, add complexity when the problem demands it

Next capsule: Project — you'll build the first version of the AI Research Assistant using the Functional API. It's the start of a project that will keep evolving across the next 6 modules.

Additional resources

1. LangGraph Functional API Conceptual Guide — Official documentation for @entrypoint and @task
2. LangGraph Low-Level Concepts — StateGraph, nodes, edges and state concepts
3. LangGraph Checkpointing Guide — How checkpointing works in both APIs
4. LangGraph How-To: Subgraphs — Composing graphs inside graphs
5. LangGraph Tutorials — Official tutorials covering both APIs
6. YAGNI Principle (Martin Fowler) — The design principle that guides when to add complexity

Module 6 — LangChain & LangGraph: From Chains to Agents