Native Control Flow

Capsule overview

This is the Functional API's killer feature: you use standard Python control flow — while, if/else, for, try/except — to steer your workflow's execution. No new DSL, no conditional edges, no special routing methods. Just Python.

In the Graph API (Module 5), to make an agent loop until the model stopped asking for tools, you needed add_conditional_edges with a router function. To route by intent type, you needed another add_conditional_edges. To handle errors, the logic lived inside each individual node.

With the Functional API, all of that becomes a while True, an if/else, and a try/except. The code reads like plain Python — because it is plain Python. LangGraph takes care of checkpointing and streaming underneath, but you write control logic the way you always have.

This capsule will walk you through every control-flow pattern, comparing them side by side with the Graph API so you understand exactly how much code you save (and when the Graph API is still the better option).

while loops: the agent loop

The most important pattern in agents is the ReAct loop: the model generates a response, if it includes tool calls you execute them, you feed the results back to the model, and you repeat until the model answers without asking for tools.

With the Functional API, this is a while True:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    data = {"Madrid": "Sunny, 22°C", "Mexico City": "Cloudy, 18°C", "Buenos Aires": "Rainy, 15°C"}
    return data.get(city, f"No data for {city}")

@tool
def get_population(city: str) -> str:
    """Gets the population of a city."""
    data = {"Madrid": "3.2 million", "Mexico City": "9.2 million", "Buenos Aires": "3.1 million"}
    return data.get(city, f"No data for {city}")

tools = [get_weather, get_population]
tool_map = {t.name: t for t in tools}

@task
def call_model(messages: list) -> object:
    """Calls the model with the available tools."""
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.bind_tools(tools).invoke(messages)

@task
def execute_tools(tool_calls: list) -> list:
    """Runs the tool calls and returns the results."""
    results = []
    for tc in tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return results

@entrypoint()
def agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    while True:
        response = call_model(messages).result()
        
        if not response.tool_calls:
            return response.content
        
        tool_results = execute_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results

result = agent.invoke("What's the weather in Madrid and how many people live there?")
print(result)
# Expected output: In Madrid the weather is sunny at 22°C and it has a
# population of roughly 3.2 million people.

Read the while True as if it were pseudocode:

1. Call the model
2. If there are no tool calls → return the response (we leave the loop)
3. If there are tool calls → run them and add the results to the messages
4. Repeat

That's the complete ReAct loop. No edges, no conditional edges, no router functions. Just a while and an if.

if/else: routing by intent

A common case: classify the user's intent and route to different tasks depending on the result.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def classify_intent(message: str) -> str:
    """Classifies the message's intent."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify the intent of the following message in exactly one word: "
        f"'code', 'question', or 'creative'.\n\nMessage: {message}\n\nIntent:"
    )
    return response.content.strip().lower()

@task
def handle_code(message: str) -> str:
    """Generates code based on the message."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Generate Python code for: {message}")
    return f"[Code Mode]\n{response.content}"

@task
def handle_question(message: str) -> str:
    """Answers a question."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Answer concisely: {message}")
    return f"[Q&A Mode]\n{response.content}"

@task
def handle_creative(message: str) -> str:
    """Generates creative content."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Write something creative about: {message}")
    return f"[Creative Mode]\n{response.content}"

@entrypoint()
def router_agent(message: str) -> str:
    intent = classify_intent(message).result()
    
    if intent == "code":
        return handle_code(message).result()
    elif intent == "creative":
        return handle_creative(message).result()
    else:
        return handle_question(message).result()

print(router_agent.invoke("Write a function that sorts a list"))
# Expected output: [Code Mode]
# ```python
# def sort_list(lst):
#     return sorted(lst)
# ...

print(router_agent.invoke("How many planets does the solar system have?"))
# Expected output: [Q&A Mode]
# The solar system has 8 planets...

print(router_agent.invoke("Write a haiku about programming"))
# Expected output: [Creative Mode]
# Code flowing softly / between lines of quiet dusk / bugs fade into light

Routing is a standard if/elif/else. If you need to add a new route, you add an elif. You don't need to touch edges or write new router functions.

for loops: iterating over multiple items

When you need to process a collection of items, a for loop is the natural answer:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def analyze_sentiment(text: str) -> dict:
    """Analyzes the sentiment of a text."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify the sentiment of this text as 'positive', 'negative', or 'neutral'. "
        f"Reply with the classification only.\n\nText: {text}"
    )
    sentiment = response.content.strip().lower()
    return {"text": text[:50], "sentiment": sentiment}

@entrypoint()
def batch_sentiment(reviews: list[str]) -> list[dict]:
    results = []
    for review in reviews:
        analysis = analyze_sentiment(review).result()
        results.append(analysis)
    return results

reviews = [
    "This product is amazing, it exceeded my expectations",
    "Awful service, I'm never buying here again",
    "The shipment arrived on time, everything is fine",
    "The quality is horrible, it broke on first use",
]

output = batch_sentiment.invoke(reviews)
for item in output:
    print(f"  {item['sentiment']:>10} → {item['text']}")
# Expected output:
#    positive → This product is amazing, it exceeded my expectati
#    negative → Awful service, I'm never buying here again
#     neutral → The shipment arrived on time, everything is fine
#    negative → The quality is horrible, it broke on first use

And with parallel execution using the Futures pattern:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def analyze_sentiment(text: str) -> dict:
    """Analyzes the sentiment of a text."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify the sentiment as 'positive', 'negative', or 'neutral'. "
        f"Classification only.\n\nText: {text}"
    )
    return {"text": text[:50], "sentiment": response.content.strip().lower()}

@entrypoint()
def parallel_batch_sentiment(reviews: list[str]) -> list[dict]:
    futures = [analyze_sentiment(review) for review in reviews]
    results = [f.result() for f in futures]
    return results

reviews = [
    "This product is amazing, it exceeded my expectations",
    "Awful service, I'm never buying here again",
    "The shipment arrived on time, everything is fine",
]

output = parallel_batch_sentiment.invoke(reviews)
for item in output:
    print(f"  {item['sentiment']:>10} → {item['text']}")
# Expected output:
#    positive → This product is amazing, it exceeded my expectati
#    negative → Awful service, I'm never buying here again
#     neutral → The shipment arrived on time, everything is fine

The difference: in the sequential version, each analysis waits for the previous one. In the parallel version, they all launch at once and the results are collected afterwards.

try/except: error handling with graceful degradation

In production, APIs fail. Models return unexpected responses. Input data is corrupted. try/except lets you handle these cases without your entire workflow collapsing:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def fetch_primary_data(query: str) -> str:
    """Fetches data from the primary source."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Answer in one sentence: {query}")
    return response.content

@task
def fetch_fallback_data(query: str) -> str:
    """Fallback source — a generic response."""
    return f"No specific information found about '{query}'. Check additional sources."

@task
def generate_response(data: str, source: str) -> str:
    """Generates the final response, noting the source."""
    return f"[Source: {source}] {data}"

@entrypoint()
def resilient_agent(query: str) -> str:
    try:
        data = fetch_primary_data(query).result()
        source = "primary"
    except Exception as e:
        print(f"  Primary source failed: {e}. Using the fallback...")
        data = fetch_fallback_data(query).result()
        source = "fallback"
    
    response = generate_response(data, source).result()
    return response

result = resilient_agent.invoke("What is quantum computing?")
print(result)
# Expected output: [Source: primary] Quantum computing is a computing paradigm
# that uses qubits instead of classical bits...

The pattern is a classic: try the primary source, and if it fails, use the fallback. In the Graph API, this kind of logic required handling exceptions inside each individual node, with no clean way to take an alternative path from the graph level.

Combining try/except with a loop for retry

from dotenv import load_dotenv
load_dotenv()

import time
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def unreliable_api_call(query: str) -> str:
    """Simulates an API that can fail."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Answer briefly: {query}")
    return response.content

@entrypoint()
def retry_agent(query: str) -> str:
    max_attempts = 3
    
    for attempt in range(max_attempts):
        try:
            result = unreliable_api_call(query).result()
            return result
        except Exception as e:
            if attempt < max_attempts - 1:
                wait = 2 ** attempt
                print(f"  Attempt {attempt + 1} failed: {e}. Waiting {wait}s...")
                time.sleep(wait)
            else:
                return f"Error: Could not get a response after {max_attempts} attempts."

result = retry_agent.invoke("What is machine learning?")
print(result)
# Expected output: Machine learning is a branch of artificial intelligence
# that lets systems learn from data without being explicitly programmed.

A for loop with try/except and exponential backoff. Pure Python control flow replacing what a graph-based system would need retry nodes, conditional edges back to the failed node, and state to track attempts for.

Combining control flow: while + if/else + try/except

The real power shows up when you combine patterns. Here's a full agent that uses all three:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool

@tool
def search_docs(query: str) -> str:
    """Searches the internal documentation."""
    docs = {
        "pricing": "Basic plan: $10/mo. Pro plan: $30/mo. Enterprise: contact sales.",
        "refund": "Refund policy: 30 days, no questions asked.",
        "hours": "Support hours: Monday to Friday, 9:00 to 18:00 CT.",
    }
    for key, value in docs.items():
        if key in query.lower():
            return value
    return "No relevant information found in the documentation."

@tool
def escalate_to_human(reason: str) -> str:
    """Escalates the case to a human agent."""
    return f"Case escalated. Reason: {reason}. An agent will contact you within 24h."

tools = [search_docs, escalate_to_human]
tool_map = {t.name: t for t in tools}

@task
def call_support_model(messages: list) -> object:
    """Calls the support model."""
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content=(
        "You are a support agent. Use search_docs to look up information. "
        "If you can't resolve it, use escalate_to_human. Respond in English."
    ))
    return model.bind_tools(tools).invoke([system] + messages)

@task
def run_tools(tool_calls: list) -> list:
    """Runs the tool calls."""
    results = []
    for tc in tool_calls:
        try:
            output = tool_map[tc["name"]].invoke(tc["args"])
            results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
        except Exception as e:
            results.append(ToolMessage(
                content=f"Error running {tc['name']}: {e}",
                tool_call_id=tc["id"]
            ))
    return results

@entrypoint()
def support_agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    max_iterations = 5
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        try:
            response = call_support_model(messages).result()
        except Exception as e:
            return f"System error: {e}. Please try again later."
        
        if not response.tool_calls:
            return response.content
        
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Iteration limit reached. Escalating to a human agent."

print(support_agent.invoke("How much does the pro plan cost?"))
# Expected output: The pro plan costs $30 a month. Is there anything else I
# can help you with?

print(support_agent.invoke("I want to talk to a human, my order is lost"))
# Expected output: I've escalated your case to a human agent. The reason on
# record is that your order is lost. They'll contact you within 24 hours.

This agent combines:

• while with an iteration limit (avoids infinite loops)
• if to decide whether it's done or needs to run tools
• try/except to handle errors from the model and from tools

Side-by-side comparison: Graph API vs Functional API

Let's look at the same ReAct loop implemented with both APIs.

Graph API (StateGraph + conditional edges)

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display

@tool
def get_weather(city: str) -> str:
    """Gets the weather for a city."""
    return f"Sunny, 22°C in {city}"

tools = [get_weather]
tool_map = {t.name: t for t in tools}

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

def call_model(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def run_tools(state: State) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue(state: State) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

graph_builder = StateGraph(State)
graph_builder.add_node("model", call_model)
graph_builder.add_node("tools", run_tools)
graph_builder.add_edge(START, "model")
graph_builder.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph_builder.add_edge("tools", "model")

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="Weather in Madrid?")]})
print(result["messages"][-1].content)
# Expected output: The weather in Madrid is sunny with a temperature of 22°C.

Functional API (while loop)

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Gets the weather for a city."""
    return f"Sunny, 22°C in {city}"

tools = [get_weather]
tool_map = {t.name: t for t in tools}

@task
def call_model(messages: list) -> object:
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.bind_tools(tools).invoke(messages)

@task
def run_tools(tool_calls: list) -> list:
    results = []
    for tc in tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return results

@entrypoint()
def agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    while True:
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results

result = agent.invoke("Weather in Madrid?")
print(result)
# Expected output: The weather in Madrid is sunny with a temperature of 22°C.

Key differences

Equivalence table: Python ↔ Graph API

When Python control flow ISN'T enough

The Functional API doesn't replace the Graph API in every case. There are scenarios where the Graph API is clearly superior:

1. Complex topologies with many parallel nodes

If your workflow has 10+ nodes with crossed dependencies (A feeds C and D, B feeds D and E, C and D feed F...), the Graph API expresses those dependencies more clearly than nesting loops and conditionals.

2. Visualization is critical

The Graph API generates diagrams with draw_mermaid_png(). If your team needs to see the workflow's topology to understand it, the Graph API wins. The Functional API doesn't produce a diagram — you'd have to read the code.

3. Multiple re-entry points

If your workflow needs to "jump" from one point to another that isn't the next step (it isn't linear), the Graph API handles this naturally with edges. With Python control flow, you'd need flags and conditionals that become hard to maintain.

4. Reusable sub-workflows

The Graph API lets you compose graphs as subgraphs. A compiled StateGraph can be used as a node inside another graph. With the Functional API, composition happens at the function level (which also works, but without the state-isolation guarantees subgraphs give you).

The practical rule:

• ✅ Functional API: sequential flows, simple loops, routing with 2-4 branches, fast prototypes
• ✅ Graph API: complex topologies, visualization required, reusable sub-workflows, large teams that need diagrams

Nested control flow: loops with conditionals

Real patterns combine multiple levels of control flow. Let's look at an agent that processes a list of documents with different logic depending on the type:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def classify_document(doc: str) -> str:
    """Classifies a document by type."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify this text as 'technical', 'business', or 'legal'. "
        f"Classification only.\n\nText: {doc}"
    )
    return response.content.strip().lower()

@task
def summarize_technical(doc: str) -> str:
    """Summarizes a technical document, focusing on specifications."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Summarize the key technical specifications: {doc}")
    return f"[TECH] {response.content}"

@task
def summarize_business(doc: str) -> str:
    """Summarizes a business document, focusing on metrics."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Summarize the key metrics and KPIs: {doc}")
    return f"[BIZ] {response.content}"

@task
def summarize_legal(doc: str) -> str:
    """Summarizes a legal document, focusing on obligations."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Summarize the key legal obligations: {doc}")
    return f"[LEGAL] {response.content}"

@entrypoint()
def process_documents(documents: list[str]) -> list[str]:
    summaries = []
    
    for doc in documents:
        doc_type = classify_document(doc).result()
        
        if doc_type == "technical":
            summary = summarize_technical(doc).result()
        elif doc_type == "business":
            summary = summarize_business(doc).result()
        elif doc_type == "legal":
            summary = summarize_legal(doc).result()
        else:
            summary = f"[UNKNOWN] Unrecognized type: {doc_type}"
        
        summaries.append(summary)
    
    return summaries

docs = [
    "The server runs PostgreSQL 16 with 32GB RAM and async replication.",
    "Q3 revenue: $2.4M, up 15% YoY. CAC reduced to $45.",
    "The contractor is obligated to deliver the software before December 31.",
]

results = process_documents.invoke(docs)
for r in results:
    print(r)
# Expected output:
# [TECH] The server uses PostgreSQL 16 with 32GB of RAM...
# [BIZ] Q3 revenue came in at $2.4M, a 15% increase...
# [LEGAL] The contractor has an obligation to deliver...

A for loop iterating over documents, with an if/elif/else inside routing to the right task by type. In the Graph API this would need a classification node, conditional edges to three different summarization nodes, and then edges back to an accumulator node. More nodes, more edges, more code.

Troubleshooting

Problem 1: "Infinite loop — the agent never finishes"

Symptom: The workflow runs forever. Cause: The while True has no exit condition, or the model keeps asking for tool calls on every iteration. Fix: Add an iteration limit:

@entrypoint()
def safe_agent(message: str) -> str:
    messages = [HumanMessage(content=message)]
    max_iterations = 10
    
    for i in range(max_iterations):
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Iteration limit reached."

Problem 2: "The if/else doesn't route correctly"

Symptom: The classification returns strings with whitespace or unexpected capitalization, and none of the if branches fire. Cause: The model returns "Technical\n" or " Code " instead of "technical". Fix: Normalize the model's response before comparing:

intent = classify_intent(message).result()
intent = intent.strip().lower()

if intent == "code":
    ...

Problem 3: "try/except catches errors it shouldn't"

Symptom: The try/except hides legitimate bugs (like a TypeError or KeyError). Cause: A generic except Exception catches everything. Fix: Catch only the exceptions you expect:

from langchain_core.exceptions import OutputParserException

try:
    result = call_api(query).result()
except (ConnectionError, TimeoutError) as e:
    result = fallback(query).result()
# TypeError, KeyError, etc. propagate normally

Problem 4: "The for loop is very slow with many items"

Symptom: Processing 20 items takes a long time because each one waits for the previous. Cause: You're calling .result() inside the loop on every iteration. Fix: Use the parallel Futures pattern:

# ❌ Sequential
for item in items:
    result = process(item).result()  # Blocks on every iteration

# ✅ Parallel
futures = [process(item) for item in items]
results = [f.result() for f in futures]

Problem 5: "I want to visualize my workflow but there's no draw_mermaid_png()"

Symptom: The Functional API has no visualization method. Cause: The Functional API doesn't build an explicit graph — the flow lives in your Python code. Fix: To understand a functional workflow's topology, read the @entrypoint's code. If you need a diagram, consider using the Graph API for complex workflows, or document the flow manually with a Mermaid diagram.

Exercises

Exercise 1: A while loop with a counter (Easy)

Create an agent that uses a while loop to "refine" an answer. On each iteration, the model improves its previous answer. The loop ends when the model says "FINAL:" at the start of its response, or after 3 iterations.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def refine(prompt: str, previous: str, iteration: int) -> str:
    """Refines a previous answer."""
    model = init_chat_model("openai:gpt-4.1-mini")
    if previous:
        msg = (
            f"Your previous answer was: '{previous}'. "
            f"Improve it (iteration {iteration}). "
            f"If it's already good, start your answer with 'FINAL:'. "
            f"Original question: {prompt}"
        )
    else:
        msg = f"Answer briefly: {prompt}"
    return model.invoke(msg).content

@entrypoint()
def refinement_agent(question: str) -> str:
    previous = ""
    max_iterations = 3
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        response = refine(question, previous, iteration).result()
        
        if response.startswith("FINAL:"):
            return response[6:].strip()
        
        previous = response
    
    return previous

result = refinement_agent.invoke("What is Docker in one sentence?")
print(result)
# Expected output: Docker is a containerization platform that packages
# applications and their dependencies into portable, isolated containers.

Exercise 2: if/else routing to 3 specialists (Easy)

Create a workflow that classifies a user message as "math", "history", or "science", and routes it to the right specialist (each one is a different @task that answers with an identifying prefix).

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def classify(message: str) -> str:
    """Classifies the message as math, history, or science."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Classify this question as 'math', 'history', or 'science'. "
        f"One word only.\n\nQuestion: {message}"
    )
    return response.content.strip().lower()

@task
def math_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"As a mathematics expert, answer: {question}")
    return f"[Mathematics] {response.content}"

@task
def history_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"As a history expert, answer: {question}")
    return f"[History] {response.content}"

@task
def science_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"As a science expert, answer: {question}")
    return f"[Science] {response.content}"

@entrypoint()
def specialist_router(question: str) -> str:
    category = classify(question).result()
    
    if category == "math":
        return math_expert(question).result()
    elif category == "history":
        return history_expert(question).result()
    else:
        return science_expert(question).result()

print(specialist_router.invoke("What is the integral of x²?"))
# Expected output: [Mathematics] The integral of x² is (x³)/3 + C...

print(specialist_router.invoke("Who was Napoleon?"))
# Expected output: [History] Napoleon Bonaparte was a military leader...

print(specialist_router.invoke("How does photosynthesis work?"))
# Expected output: [Science] Photosynthesis is the process by which...

Exercise 3: A for loop with parallel processing (Medium)

Take a list of 4 cities. For each city, a @task fetches simulated information. Run them all in parallel and return a combined report.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def get_city_info(city: str) -> dict:
    """Fetches simulated information about a city."""
    data = {
        "Madrid": {"population": "3.2M", "country": "Spain", "temp": "22°C"},
        "Mexico City": {"population": "9.2M", "country": "Mexico", "temp": "18°C"},
        "Buenos Aires": {"population": "3.1M", "country": "Argentina", "temp": "15°C"},
        "Bogotá": {"population": "7.4M", "country": "Colombia", "temp": "14°C"},
    }
    info = data.get(city, {"population": "N/A", "country": "N/A", "temp": "N/A"})
    return {"city": city, **info}

@entrypoint()
def city_report(cities: list[str]) -> str:
    futures = [get_city_info(city) for city in cities]
    results = [f.result() for f in futures]
    
    report = "City report:\n"
    for info in results:
        report += (
            f"  • {info['city']} ({info['country']}): "
            f"{info['population']} inhabitants, {info['temp']}\n"
        )
    return report

output = city_report.invoke(["Madrid", "Mexico City", "Buenos Aires", "Bogotá"])
print(output)
# Expected output:
# City report:
#   • Madrid (Spain): 3.2M inhabitants, 22°C
#   • Mexico City (Mexico): 9.2M inhabitants, 18°C
#   • Buenos Aires (Argentina): 3.1M inhabitants, 15°C
#   • Bogotá (Colombia): 7.4M inhabitants, 14°C

Exercise 4: try/except with a 3-level fallback (Medium)

Create a workflow that tries 3 sources in order: primary, secondary, and local fallback. If a source fails, try the next one. If all of them fail, return an error message.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

@task
def query_primary(question: str) -> str:
    """Primary source — the main model."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Answer concisely: {question}")
    return f"[Primary] {response.content}"

@task
def query_secondary(question: str) -> str:
    """Secondary source — an alternative model."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Answer in simple terms: {question}")
    return f"[Secondary] {response.content}"

@task
def query_local_cache(question: str) -> str:
    """Local source — cached answers."""
    cache = {
        "python": "Python is a high-level programming language.",
        "javascript": "JavaScript is a language for web development.",
    }
    for key, value in cache.items():
        if key in question.lower():
            return f"[Cache] {value}"
    return f"[Cache] No cached answer for: {question}"

@entrypoint()
def resilient_query(question: str) -> str:
    sources = [
        ("Primary", query_primary),
        ("Secondary", query_secondary),
        ("Cache", query_local_cache),
    ]
    
    for source_name, source_fn in sources:
        try:
            result = source_fn(question).result()
            return result
        except Exception as e:
            print(f"  {source_name} failed: {e}")
            continue
    
    return "Error: All sources failed. Try again later."

result = resilient_query.invoke("What is Python?")
print(result)
# Expected output: [Primary] Python is a high-level, interpreted,
# general-purpose programming language...

Exercise 5: A complete ReAct agent with while + if + try/except (Advanced)

Build a technical support agent with 2 tools (check_status and restart_service). The agent uses a while loop (max 5 iterations), if/else to decide whether to run tools or answer, and try/except to handle errors during tool execution.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool

@tool
def check_status(service: str) -> str:
    """Checks a service's status."""
    statuses = {
        "api": "running (99.9% uptime)",
        "database": "degraded (high latency: 500ms)",
        "cache": "down (last restart: 2h ago)",
    }
    return statuses.get(service.lower(), f"Service '{service}' not found")

@tool
def restart_service(service: str) -> str:
    """Restarts a service."""
    if service.lower() == "cache":
        return f"Service '{service}' restarted successfully. Status: running."
    return f"Service '{service}' doesn't need a restart. Current status: running."

tools = [check_status, restart_service]
tool_map = {t.name: t for t in tools}

@task
def call_support_model(messages: list) -> object:
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content=(
        "You are a technical support agent. "
        "Use check_status to check services and restart_service to restart them. "
        "Respond in English."
    ))
    return model.bind_tools(tools).invoke([system] + messages)

@task
def execute_tool_calls(tool_calls: list) -> list:
    results = []
    for tc in tool_calls:
        try:
            output = tool_map[tc["name"]].invoke(tc["args"])
            results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
        except Exception as e:
            results.append(ToolMessage(
                content=f"Error: {e}", tool_call_id=tc["id"]
            ))
    return results

@entrypoint()
def tech_support(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    for iteration in range(5):
        try:
            response = call_support_model(messages).result()
        except Exception as e:
            return f"System error on iteration {iteration + 1}: {e}"
        
        if not response.tool_calls:
            return response.content
        
        tool_results = execute_tool_calls(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Iteration limit reached. Escalating to an engineer."

print(tech_support.invoke("The cache is down, can you check it and restart it?"))
# Expected output: I checked the cache's status and it was down. I restarted it
# successfully and it's now running correctly.

Exercise 6: The same problem with the Graph API and the Functional API (Advanced)

Implement a classifier-router that takes a message, classifies it (question vs command), and routes it to the right task. Do it twice: once with StateGraph + add_conditional_edges, and once with @entrypoint + if/else. Compare the amount of code.

from dotenv import load_dotenv
load_dotenv()

# ========================================
# VERSION 1: Graph API (StateGraph)
# ========================================
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str

def classify(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["?", "what", "how", "when"]):
        return {"intent": "question"}
    return {"intent": "command"}

def handle_question(state: State) -> dict:
    return {"messages": [AIMessage(content="[Q&A] Processing your question...")]}

def handle_command(state: State) -> dict:
    return {"messages": [AIMessage(content="[CMD] Running your command...")]}

def route(state: State) -> str:
    if state["intent"] == "question":
        return "question"
    return "command"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("question", handle_question)
builder.add_node("command", handle_command)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"question": "question", "command": "command"})
builder.add_edge("question", END)
builder.add_edge("command", END)

graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="How does Python work?")], "intent": ""})
print(f"Graph API: {result['messages'][-1].content}")
# Graph API: [Q&A] Processing your question...

# ========================================
# VERSION 2: Functional API
# ========================================
from langgraph.func import entrypoint, task

@task
def func_classify(message: str) -> str:
    text = message.lower()
    if any(w in text for w in ["?", "what", "how", "when"]):
        return "question"
    return "command"

@task
def func_handle_question(message: str) -> str:
    return f"[Q&A] Processing your question..."

@task
def func_handle_command(message: str) -> str:
    return f"[CMD] Running your command..."

@entrypoint()
def func_router(message: str) -> str:
    intent = func_classify(message).result()
    
    if intent == "question":
        return func_handle_question(message).result()
    else:
        return func_handle_command(message).result()

result = func_router.invoke("How does Python work?")
print(f"Functional API: {result}")
# Functional API: [Q&A] Processing your question...

Summary

In this capsule you learned:

• The Functional API uses standard Python control flow (while, if/else, for, try/except) instead of edges and conditional edges
• while loops replace conditional edges back to the same node — perfect for an agent's ReAct loop
• if/else replaces add_conditional_edges — direct routing without separate router functions
• for loops let you iterate over dynamic collections — with the option of parallel execution using Futures
• try/except enables graceful degradation — multi-level fallbacks without complicating the graph's topology
• The patterns combine: a real agent uses while + if + try/except inside the same @entrypoint
• Graph API vs Functional API: the Functional API wins on sequential flows and simple routing; the Graph API wins on complex topologies, visualization, and reusable sub-workflows
• The code reads like plain Python — that's the value proposition. LangGraph handles checkpointing and streaming underneath, but the control logic is yours

Next capsule: you'll learn to combine both APIs — using a StateGraph as a component inside an @entrypoint, and the other way around — to get the best of both worlds.

Additional resources

1. LangGraph Functional API — Conceptual Guide — Official docs on control flow in the Functional API
2. LangGraph Functional API — How-To Guide — Step-by-step tutorial with while, if/else, and for examples
3. Graph API vs Functional API — Official comparison of the two APIs
4. ReAct Pattern — LangGraph — The ReAct pattern implemented in LangGraph
5. Conditional Edges — Graph API — Conditional edges reference for the comparison
6. LangGraph Streaming — How streaming works with native control flow
7. Python Control Flow — Official Tutorial — Reference for while, if/else, for, try/except in Python

Module 6 — LangChain & LangGraph: From Chains to Agents