Patterns with the Functional API

Capsule overview

You already understand the Functional API and how it compares to the Graph API. Now let's get practical: reusable patterns — recipes that solve common problems naturally with @entrypoint and @task. These are the patterns you'll copy and adapt in your real projects.

Each pattern solves a type of architectural problem: tool execution loops, multi-step reasoning, parallelism with Futures, structured data extraction, and error handling with fallback. They're the building blocks that make the Functional API shine — flows expressed as clean Python with durability and checkpointing superpowers.

Pattern 1: Tool execution loop

The most fundamental pattern for agents: the model decides whether it needs to call tools, runs them, and goes back to the model. It's the ReAct loop implemented with a while and @task.

Why it works well with the Functional API

In the Graph API, this pattern requires a conditional edge that checks whether there are tool calls, plus a circular edge back to the model node. With the Functional API, it's a while loop — exactly how you'd think about it before you ever heard of LangGraph.

Full implementation

from dotenv import load_dotenv
load_dotenv()

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

@tool
def search_web(query: str) -> str:
    """Searches the web for information about a topic."""
    return f"Results for '{query}': Python 3.12 includes improvements in typing, performance and f-strings."

@tool
def get_date() -> str:
    """Gets the current date."""
    return "March 8, 2026"

tools_list = [search_web, get_date]
tool_map = {t.name: t for t in tools_list}
model = init_chat_model("openai:gpt-4.1-mini").bind_tools(tools_list)

@task
def call_model(messages: list[BaseMessage]):
    return model.invoke(messages)

@task
def execute_tool(tool_call: dict):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def react_agent(user_message: str) -> str:
    messages = [
        SystemMessage(content="You are a research assistant. Use the available tools."),
        HumanMessage(content=user_message),
    ]

    while True:
        response = call_model(messages).result()

        if not response.tool_calls:
            return response.content

        tool_futures = [execute_tool(tc) for tc in response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]

        messages = add_messages(messages, [response, *tool_results])

result = react_agent.invoke("What's new in Python 3.12? And what's today's date?")
print(result)
# Expected output: Python 3.12 includes improvements in typing, performance and f-strings.
# Today's date is March 8, 2026.

Anatomy of the pattern

1. Setup: Model with tools bound, tool map keyed by name
2. Loop: while True with an implicit break in the return
3. Decision: if not response.tool_calls: return — if the model doesn't ask for tools, we're done
4. Parallel execution: The tools are launched as Futures and collected afterwards
5. Accumulation: add_messages updates the conversation history

The tools run in parallel automatically — if the model asks for search_web and get_date in the same response, both are launched without waiting for one to finish before starting the other.

Pattern 2: Multi-step reasoning

Break a complex problem into steps: analyze → reason → synthesize. Each step as an independent @task with its own checkpoint.

Why it works well with the Functional API

Multi-step reasoning is inherently sequential: each step depends on the previous one. The Functional API expresses it as a sequence of @task calls — no nodes, no edges, just functions that call each other one after another.

Full implementation

from dotenv import load_dotenv
load_dotenv()

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

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

@task
def decompose_question(question: str) -> list:
    response = model.invoke(
        f"Break this complex question into 2-3 simpler sub-questions. "
        f"Answer ONLY with the sub-questions, one per line.\n\n"
        f"Question: {question}"
    )
    sub_questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return sub_questions

@task
def reason_about(sub_question: str) -> str:
    response = model.invoke(
        f"Answer this question concisely and technically (2-3 sentences):\n{sub_question}"
    )
    return response.content

@task
def synthesize(question: str, partial_answers: list) -> str:
    context = "\n".join(f"- {a}" for a in partial_answers)
    response = model.invoke(
        f"Using these partial answers, produce a complete and coherent answer.\n\n"
        f"Original question: {question}\n\n"
        f"Partial answers:\n{context}"
    )
    return response.content

@entrypoint()
def chain_of_thought(question: str) -> dict:
    sub_questions = decompose_question(question).result()

    partial_answers = []
    for sq in sub_questions:
        answer = reason_about(sq).result()
        partial_answers.append(answer)

    final_answer = synthesize(question, partial_answers).result()

    return {
        "question": question,
        "sub_questions": sub_questions,
        "partial_answers": partial_answers,
        "final_answer": final_answer,
    }

result = chain_of_thought.invoke(
    "Why did transformers replace RNNs in natural language processing?"
)
print(f"Sub-questions: {len(result['sub_questions'])}")
for i, sq in enumerate(result["sub_questions"], 1):
    print(f"  {i}. {sq}")
print(f"\nFinal answer: {result['final_answer'][:150]}...")
# Expected output:
# Sub-questions: 2-3
#   1. What were the limitations of RNNs?
#   2. What advantages does the transformer architecture introduce?
#   3. What empirical evidence showed the superiority of transformers?
# Final answer: Transformers replaced RNNs mainly because...

Automatic checkpointing

Every @task saves its result to a checkpoint. If the workflow fails at synthesize, on resume it doesn't re-run decompose_question or the reason_about calls that already completed — it recovers their results from the checkpoint. In multi-step reasoning with 5 sub-questions where each one requires an LLM call, this saves time and tokens.

Pattern 3: Parallel task execution with Futures

Launch multiple tasks without waiting for each one to finish, and collect all the results at the end. The parallelism pattern that makes the Functional API practical for I/O-bound workloads.

Why it works well with the Functional API

Calling a @task returns a Future immediately — the task runs in the background. If you call 3 tasks without .result(), all 3 can run in parallel. When you finally call .result(), you block until the result is ready. This pattern is natural for searching multiple sources, processing multiple documents, or any I/O-bound operation.

Full implementation

from dotenv import load_dotenv
load_dotenv()

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

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

@task
def search_wikipedia(topic: str) -> str:
    response = model.invoke(
        f"Simulate a Wikipedia search about '{topic}'. "
        f"Generate an informative paragraph as if it were a Wikipedia article."
    )
    return f"[Wikipedia] {response.content}"

@task
def search_arxiv(topic: str) -> str:
    response = model.invoke(
        f"Simulate an arXiv search about '{topic}'. "
        f"Generate a technical summary as if it were a paper abstract."
    )
    return f"[arXiv] {response.content}"

@task
def search_news(topic: str) -> str:
    response = model.invoke(
        f"Simulate a search for recent news about '{topic}'. "
        f"Generate a short journalistic summary."
    )
    return f"[News] {response.content}"

@task
def synthesize_results(topic: str, sources: list) -> str:
    context = "\n\n".join(sources)
    response = model.invoke(
        f"Synthesize these sources into a 3-4 sentence executive summary about '{topic}':\n\n{context}"
    )
    return response.content

@entrypoint()
def parallel_research(topic: str) -> dict:
    wiki_future = search_wikipedia(topic)
    arxiv_future = search_arxiv(topic)
    news_future = search_news(topic)

    wiki = wiki_future.result()
    arxiv = arxiv_future.result()
    news = news_future.result()

    summary = synthesize_results(topic, [wiki, arxiv, news]).result()

    return {
        "topic": topic,
        "sources": [wiki, arxiv, news],
        "summary": summary,
    }

result = parallel_research.invoke("large language models fine-tuning")
print(f"Sources collected: {len(result['sources'])}")
for source in result["sources"]:
    print(f"  {source[:80]}...")
print(f"\nSummary: {result['summary'][:150]}...")
# Expected output:
# Sources collected: 3
#   [Wikipedia] Large language models (LLMs) are neural networks...
#   [arXiv] We present a comparative analysis of fine-tuning techniques...
#   [News] Technology companies are adopting fine-tuning techniques...
# Summary: Fine-tuning of LLMs has evolved from...

The mechanics of Futures

wiki_future = search_wikipedia(topic)    # Returns a Future (task is launched)
arxiv_future = search_arxiv(topic)       # Returns a Future (task is launched)
news_future = search_news(topic)         # Returns a Future (task is launched)
# All 3 tasks are running in parallel

wiki = wiki_future.result()    # Blocks until wiki finishes
arxiv = arxiv_future.result()  # Blocks until arxiv finishes
news = news_future.result()    # Blocks until news finishes

The key difference: if you call .result() right after each task, you lose the parallelism — each task waits for the previous one. The pattern is: launch all the tasks first, collect all the results afterwards.

# ❌ Sequential (no parallelism)
wiki = search_wikipedia(topic).result()    # Waits
arxiv = search_arxiv(topic).result()       # Waits
news = search_news(topic).result()         # Waits

# ✅ Parallel (launch all, collect afterwards)
wiki_fut = search_wikipedia(topic)         # Launch
arxiv_fut = search_arxiv(topic)            # Launch
news_fut = search_news(topic)              # Launch
wiki = wiki_fut.result()                   # Collect
arxiv = arxiv_fut.result()                 # Collect
news = news_fut.result()                   # Collect

Pattern 4: Structured output extraction

Use Pydantic + with_structured_output() inside a @task to extract structured data from free text. It guarantees that the LLM's output has exactly the structure your code expects.

Full implementation

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

class ProductReview(BaseModel):
    sentiment: str = Field(description="'positive', 'negative', or 'neutral'")
    score: float = Field(description="Score from 0.0 to 1.0")
    key_points: list[str] = Field(description="List of key points mentioned")
    recommendation: str = Field(description="One-sentence summary")

model = init_chat_model("openai:gpt-4.1-mini")
structured_model = model.with_structured_output(ProductReview)

@task
def analyze_review(review_text: str) -> dict:
    result = structured_model.invoke(
        f"Analyze this product review and extract structured information:\n\n{review_text}"
    )
    return result.model_dump()

@task
def aggregate_reviews(analyses: list) -> dict:
    avg_score = sum(a["score"] for a in analyses) / len(analyses)
    all_points = []
    for a in analyses:
        all_points.extend(a["key_points"])

    sentiment_counts = {}
    for a in analyses:
        s = a["sentiment"]
        sentiment_counts[s] = sentiment_counts.get(s, 0) + 1

    return {
        "total_reviews": len(analyses),
        "average_score": round(avg_score, 2),
        "sentiment_distribution": sentiment_counts,
        "all_key_points": all_points,
    }

@entrypoint()
def review_analyzer(reviews: list) -> dict:
    analysis_futures = [analyze_review(r) for r in reviews]
    analyses = [fut.result() for fut in analysis_futures]
    summary = aggregate_reviews(analyses).result()
    return summary

reviews = [
    "Excellent product, the battery lasts all day. The camera is amazing. Highly recommended.",
    "So-so. The design is nice but the performance doesn't convince me. It gets very hot.",
    "Terrible after-sales service. The product arrived defective and they gave me no solution.",
]

result = review_analyzer.invoke(reviews)
print(f"Reviews analyzed: {result['total_reviews']}")
print(f"Average score: {result['average_score']}")
print(f"Sentiments: {result['sentiment_distribution']}")
print(f"Key points: {result['all_key_points'][:3]}")
# Expected output:
# Reviews analyzed: 3
# Average score: 0.5
# Sentiments: {'positive': 1, 'neutral': 1, 'negative': 1}
# Key points: ['Long-lasting battery', 'Quality camera', 'Gets very hot']

Why .model_dump() inside the @task

A @task needs to return JSON-serializable values for checkpointing. LangGraph doesn't automatically serialize the Pydantic ProductReview object, but .model_dump() converts it into a standard Python dictionary, which is serializable. It's an extra step, but it guarantees the checkpoint works correctly.

Pattern 5: Error handling with fallback

Use try/except inside the @entrypoint to handle errors gracefully: retry with a different model, degrade the response, or take an alternate path. Plain Python for resilience.

Full implementation

from dotenv import load_dotenv
load_dotenv()

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

@task
def call_primary_model(prompt: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.invoke(prompt).content

@task
def call_fallback_model(prompt: str) -> str:
    model = init_chat_model("openai:gpt-4.1-nano")
    return model.invoke(prompt).content

@task
def validate_response(response: str, min_length: int) -> dict:
    if len(response) < min_length:
        return {"valid": False, "reason": f"Response too short ({len(response)} chars, minimum {min_length})"}
    if not any(c.isalpha() for c in response):
        return {"valid": False, "reason": "Response contains no readable text"}
    return {"valid": True, "reason": "OK"}

@entrypoint()
def resilient_agent(request: dict) -> dict:
    prompt = request["prompt"]
    min_length = request.get("min_length", 50)
    max_retries = request.get("max_retries", 2)

    for attempt in range(max_retries + 1):
        try:
            if attempt == 0:
                response = call_primary_model(prompt).result()
            else:
                response = call_fallback_model(prompt).result()

            validation = validate_response(response, min_length).result()

            if validation["valid"]:
                return {
                    "response": response,
                    "attempt": attempt + 1,
                    "model_used": "primary" if attempt == 0 else "fallback",
                    "status": "success",
                }

        except Exception as e:
            if attempt == max_retries:
                return {
                    "response": f"Error after {max_retries + 1} attempts: {str(e)}",
                    "attempt": attempt + 1,
                    "model_used": "none",
                    "status": "error",
                }
            continue

    return {
        "response": response,
        "attempt": max_retries + 1,
        "model_used": "fallback",
        "status": "degraded",
    }

result = resilient_agent.invoke({
    "prompt": "Explain what machine learning is in 3 sentences.",
    "min_length": 50,
    "max_retries": 2,
})
print(f"Status: {result['status']}")
print(f"Model used: {result['model_used']}")
print(f"Attempt: {result['attempt']}")
print(f"Response: {result['response'][:120]}...")
# Expected output:
# Status: success
# Model used: primary
# Attempt: 1
# Response: Machine learning is a branch of artificial intelligence...

Anatomy of the pattern

1. Retry loop: for attempt in range(max_retries + 1) — plain Python for retry
2. Model escalation: Attempt 0 uses the primary model, retries use the fallback
3. Validation: A @task that checks the quality of the response before accepting it
4. Graceful fallback: If everything fails, it returns an informative error instead of crashing

With the Graph API, this pattern requires a validation node, a conditional edge for retry, another for fallback, and attempt-counting logic in the state. With the Functional API, it's a for loop with try/except — the pattern any Python developer already knows.

When the patterns aren't enough

These patterns cover most sequential workflows of moderate complexity. But there are clear signs that you need to migrate to the Graph API:

• ❌ You need more than 3 conditional branches converging at a merge point — the merge logic gets messy with local variables
• ❌ You want the Send API to create dynamic workers based on runtime data — the Functional API has no direct equivalent
• ❌ The workflow has sub-workflows that compose as reusable sub-graphs for other teams
• ❌ You need automatic visualization of the flow for documentation or monitoring — draw_mermaid_png() only exists in the Graph API
• ❌ You want granular streaming control per node, with stream_mode="messages" or custom stream events

If you find yourself in any of these scenarios, don't force the Functional API. It's like using a for loop when you need recursion — technically possible, but the code turns brittle and unreadable. The Graph API exists for these cases.

Troubleshooting

Problem 1: "My tasks aren't running in parallel"

Symptom: You launch 3 tasks but the total time equals the sum of all 3, not the maximum.

Cause: You're calling .result() right after each task:

# ❌ Sequential in disguise
a = task_a(x).result()
b = task_b(y).result()
c = task_c(z).result()

Fix: Separate the launch from the collection:

# ✅ Actually parallel
fut_a = task_a(x)
fut_b = task_b(y)
fut_c = task_c(z)
a, b, c = fut_a.result(), fut_b.result(), fut_c.result()

Problem 2: "My @task returns an object that won't serialize"

Symptom: Serialization error when running a workflow with a checkpointer.

Cause: The @task returns a complex Python object (a custom class, an unconverted Pydantic model, an object with methods).

Fix: Convert to serializable types before returning:

@task
def my_task(input: str) -> dict:
    result = some_complex_operation(input)
    # ❌ return result  (complex object)
    # ✅ Convert to dict/list/str
    return {"value": str(result), "score": result.score}

For Pydantic models, use .model_dump().

Problem 3: "The retry loop doesn't recover previous tasks from the checkpoint"

Symptom: When you resume a workflow after an error, the tasks that already completed get re-run.

Cause: Tasks inside a try/except that fail don't save a result. But the ones that completed successfully are recovered from the checkpoint on resume.

Fix: Make sure the error happens inside a specific task, not in logic outside of tasks. The rule: every side effect (API calls, I/O) must live inside a @task for the checkpoint to work correctly.

Problem 4: "The @entrypoint won't accept multiple arguments"

Symptom: TypeError: entrypoint function must accept exactly one positional argument.

Cause: @entrypoint only accepts one positional argument (the workflow's input). If you need to pass multiple values, use a dict:

# ❌ Multiple arguments
@entrypoint()
def my_workflow(topic: str, max_results: int) -> str:
    ...

# ✅ A single dict as input
@entrypoint()
def my_workflow(config: dict) -> str:
    topic = config["topic"]
    max_results = config.get("max_results", 5)
    ...

Exercises

Exercise 1: Basic tool loop (Easy)

Implement an agent with Pattern 1 (tool execution loop) that has a single tool: lookup_capital(country: str) -> str, which returns the capital of a country. Test it with "What is the capital of France?" and "What are the capitals of Spain, Germany and Japan?".

from dotenv import load_dotenv
load_dotenv()

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

CAPITALS = {
    "france": "Paris", "spain": "Madrid", "germany": "Berlin",
    "japan": "Tokyo", "brazil": "Brasília", "mexico": "Mexico City",
}

@tool
def lookup_capital(country: str) -> str:
    """Looks up the capital of a country."""
    return CAPITALS.get(country.lower(), f"I have no data for {country}")

tools_list = [lookup_capital]
tool_map = {t.name: t for t in tools_list}
model = init_chat_model("openai:gpt-4.1-mini").bind_tools(tools_list)

@task
def call_model(messages: list[BaseMessage]):
    return model.invoke(messages)

@task
def execute_tool(tool_call: dict):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def capital_agent(question: str) -> str:
    messages = [
        SystemMessage(content="Help the user find the capitals of countries."),
        HumanMessage(content=question),
    ]

    while True:
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_futures = [execute_tool(tc) for tc in response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]
        messages = add_messages(messages, [response, *tool_results])

print(capital_agent.invoke("What is the capital of France?"))
print(capital_agent.invoke("What are the capitals of Spain, Germany and Japan?"))
# Expected output:
# The capital of France is Paris.
# The capitals are: Spain → Madrid, Germany → Berlin, Japan → Tokyo.

Exercise 2: Chain of thought with a variable number of steps (Easy)

Adapt Pattern 2 (multi-step reasoning) so that the number of sub-questions is configurable. The input should be a dict with "question" and "num_steps". If num_steps is 2, decompose into 2 sub-questions; if it's 4, into 4.

from dotenv import load_dotenv
load_dotenv()

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

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

@task
def decompose(question: str, num_steps: int) -> list:
    response = model.invoke(
        f"Break this question into exactly {num_steps} sub-questions. "
        f"Answer ONLY with the sub-questions, one per line.\n\n"
        f"Question: {question}"
    )
    questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return questions[:num_steps]

@task
def answer_subquestion(sub_question: str) -> str:
    return model.invoke(f"Answer concisely (2-3 sentences): {sub_question}").content

@task
def synthesize(question: str, answers: list) -> str:
    context = "\n".join(f"- {a}" for a in answers)
    return model.invoke(
        f"Synthesize a complete answer.\n\nQuestion: {question}\n\nPartial answers:\n{context}"
    ).content

@entrypoint()
def configurable_cot(config: dict) -> dict:
    question = config["question"]
    num_steps = config.get("num_steps", 3)

    sub_questions = decompose(question, num_steps).result()
    answers = [answer_subquestion(sq).result() for sq in sub_questions]
    final = synthesize(question, answers).result()

    return {"sub_questions": sub_questions, "answers": answers, "final_answer": final}

result = configurable_cot.invoke({
    "question": "How does artificial intelligence impact education?",
    "num_steps": 2,
})
print(f"Sub-questions ({len(result['sub_questions'])}):")
for sq in result["sub_questions"]:
    print(f"  - {sq}")
print(f"\nAnswer: {result['final_answer'][:150]}...")
# Expected output:
# Sub-questions (2):
#   - In what ways is AI changing teaching methods?
#   - What are the risks and challenges of using AI in education?
# Answer: Artificial intelligence impacts education...

Exercise 3: Parallel search with a simulated timeout (Medium)

Implement Pattern 3 (parallel execution) with 4 search sources. One of the sources should simulate a failure (raise an exception). Use try/except to handle the failure and synthesize with the sources that did respond.

from dotenv import load_dotenv
load_dotenv()

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

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

@task
def search_source_a(topic: str) -> str:
    return f"[Source A] General information about {topic} found."

@task
def search_source_b(topic: str) -> str:
    return f"[Source B] Technical data about {topic} collected."

@task
def search_source_c(topic: str) -> str:
    raise ConnectionError(f"Source C unavailable for '{topic}'")

@task
def search_source_d(topic: str) -> str:
    return f"[Source D] Recent news about {topic} found."

@task
def synthesize(topic: str, results: list) -> str:
    context = "\n".join(results)
    return model.invoke(
        f"Synthesize these sources about '{topic}':\n\n{context}"
    ).content

@entrypoint()
def resilient_search(topic: str) -> dict:
    futures = {
        "a": search_source_a(topic),
        "b": search_source_b(topic),
        "c": search_source_c(topic),
        "d": search_source_d(topic),
    }

    results = []
    errors = []
    for name, future in futures.items():
        try:
            results.append(future.result())
        except Exception as e:
            errors.append(f"Source {name}: {str(e)}")

    summary = synthesize(topic, results).result()

    return {
        "sources_ok": len(results),
        "sources_failed": len(errors),
        "errors": errors,
        "summary": summary,
    }

result = resilient_search.invoke("quantum computing")
print(f"Successful sources: {result['sources_ok']}")
print(f"Failed sources: {result['sources_failed']}")
print(f"Errors: {result['errors']}")
print(f"Summary: {result['summary'][:100]}...")
# Expected output:
# Successful sources: 3
# Failed sources: 1
# Errors: ["Source c: Source C unavailable for 'quantum computing'"]
# Summary: Based on the available sources about quantum computing...

Exercise 4: Structured extraction of multiple entities (Medium)

Use Pattern 4 (structured output) to extract entities from a text in parallel: people, organizations and locations. Each extraction as a separate @task with its own Pydantic model. Combine the results.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class People(BaseModel):
    names: list[str] = Field(description="Names of the people mentioned")

class Organizations(BaseModel):
    names: list[str] = Field(description="Names of the organizations mentioned")

class Locations(BaseModel):
    names: list[str] = Field(description="Names of the locations mentioned")

@task
def extract_people(text: str) -> list:
    extractor = model.with_structured_output(People)
    result = extractor.invoke(f"Extract every person's name from this text:\n\n{text}")
    return result.names

@task
def extract_organizations(text: str) -> list:
    extractor = model.with_structured_output(Organizations)
    result = extractor.invoke(f"Extract every organization from this text:\n\n{text}")
    return result.names

@task
def extract_locations(text: str) -> list:
    extractor = model.with_structured_output(Locations)
    result = extractor.invoke(f"Extract every location from this text:\n\n{text}")
    return result.names

@entrypoint()
def entity_extractor(text: str) -> dict:
    people_fut = extract_people(text)
    orgs_fut = extract_organizations(text)
    locs_fut = extract_locations(text)

    return {
        "people": people_fut.result(),
        "organizations": orgs_fut.result(),
        "locations": locs_fut.result(),
    }

text = (
    "Satya Nadella, CEO of Microsoft, announced in Seattle that the company will invest "
    "in artificial intelligence together with OpenAI. Sam Altman confirmed the collaboration "
    "from the San Francisco offices."
)

result = entity_extractor.invoke(text)
print(f"People: {result['people']}")
print(f"Organizations: {result['organizations']}")
print(f"Locations: {result['locations']}")
# Expected output:
# People: ['Satya Nadella', 'Sam Altman']
# Organizations: ['Microsoft', 'OpenAI']
# Locations: ['Seattle', 'San Francisco']

Exercise 5: Complete pipeline with all the patterns (Advanced)

Combine patterns 1, 3, 4 and 5 into a single workflow: an agent that takes a topic, searches 3 sources in parallel (Pattern 3), extracts structured data from each source (Pattern 4), handles errors with fallback (Pattern 5), and uses an LLM to synthesize the final result. The tool loop (Pattern 1) doesn't apply directly here, but the retry loop does.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class SourceSummary(BaseModel):
    key_facts: list[str] = Field(description="Key facts found")
    confidence: float = Field(description="Confidence from 0.0 to 1.0")
    source_type: str = Field(description="Source type: 'academic', 'news', or 'general'")

@task
def search_source(topic: str, source_name: str, source_type: str) -> str:
    return model.invoke(
        f"Simulate a search on {source_name} about '{topic}'. "
        f"Generate an informative paragraph as if it came from a {source_type} source."
    ).content

@task
def extract_structured(text: str, source_type: str) -> dict:
    extractor = model.with_structured_output(SourceSummary)
    result = extractor.invoke(
        f"Extract key facts and assess the confidence of this text (source type '{source_type}'):\n\n{text}"
    )
    return result.model_dump()

@task
def synthesize_final(topic: str, structured_data: list) -> str:
    all_facts = []
    for sd in structured_data:
        for fact in sd["key_facts"]:
            all_facts.append(f"[{sd['source_type']}, confidence {sd['confidence']}] {fact}")
    context = "\n".join(all_facts)
    return model.invoke(
        f"Generate an executive summary about '{topic}' based on these facts:\n\n{context}"
    ).content

SOURCES = [
    {"name": "Wikipedia", "type": "general"},
    {"name": "arXiv", "type": "academic"},
    {"name": "TechCrunch", "type": "news"},
]

@entrypoint()
def research_pipeline(topic: str) -> dict:
    search_futures = [
        search_source(topic, s["name"], s["type"]) for s in SOURCES
    ]

    raw_results = []
    errors = []
    for i, fut in enumerate(search_futures):
        try:
            raw_results.append({"text": fut.result(), "type": SOURCES[i]["type"]})
        except Exception as e:
            errors.append(f"{SOURCES[i]['name']}: {str(e)}")

    if not raw_results:
        return {"error": "All sources failed", "errors": errors}

    extract_futures = [
        extract_structured(r["text"], r["type"]) for r in raw_results
    ]
    structured_data = [fut.result() for fut in extract_futures]

    summary = synthesize_final(topic, structured_data).result()

    return {
        "topic": topic,
        "sources_used": len(raw_results),
        "sources_failed": len(errors),
        "structured_data": structured_data,
        "summary": summary,
    }

result = research_pipeline.invoke("retrieval-augmented generation")
print(f"Sources: {result['sources_used']} OK, {result['sources_failed']} failed")
print(f"Facts extracted: {sum(len(sd['key_facts']) for sd in result['structured_data'])}")
print(f"Summary: {result['summary'][:150]}...")
# Expected output:
# Sources: 3 OK, 0 failed
# Facts extracted: 6-9
# Summary: RAG (Retrieval-Augmented Generation) is a technique that combines...

Exercise 6: Evaluator-optimizer loop (Advanced)

Implement a workflow that generates a summary of a text, evaluates it against specific criteria (length, clarity, completeness), and regenerates it if it doesn't pass the evaluation. Maximum 3 iterations. Use Pydantic for the structured evaluation.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class Evaluation(BaseModel):
    passes: bool = Field(description="True if the summary meets every criterion")
    clarity_score: float = Field(description="Clarity from 0.0 to 1.0")
    completeness_score: float = Field(description="Completeness from 0.0 to 1.0")
    feedback: str = Field(description="Specific feedback for improvement")

@task
def generate_summary(text: str, feedback: str = "") -> str:
    prompt = f"Summarize this text in 3-4 clear and complete sentences:\n\n{text}"
    if feedback:
        prompt += f"\n\nFeedback from the previous iteration (improve this): {feedback}"
    return model.invoke(prompt).content

@task
def evaluate_summary(original: str, summary: str) -> dict:
    evaluator = model.with_structured_output(Evaluation)
    result = evaluator.invoke(
        f"Evaluate this summary by comparing it against the original text.\n\n"
        f"Original:\n{original}\n\nSummary:\n{summary}\n\n"
        f"Criteria: clarity >= 0.7, completeness >= 0.7, length between 50-200 words."
    )
    return result.model_dump()

@entrypoint()
def summarize_with_eval(text: str) -> dict:
    feedback = ""
    max_iterations = 3

    for iteration in range(max_iterations):
        summary = generate_summary(text, feedback).result()
        evaluation = evaluate_summary(text, summary).result()

        if evaluation["passes"]:
            return {
                "summary": summary,
                "iterations": iteration + 1,
                "final_evaluation": evaluation,
                "status": "approved",
            }

        feedback = evaluation["feedback"]

    return {
        "summary": summary,
        "iterations": max_iterations,
        "final_evaluation": evaluation,
        "status": "max_iterations_reached",
    }

text = (
    "Large language models (LLMs) have transformed the field of natural language "
    "processing. Based on the transformer architecture, these models learn statistical "
    "patterns of language from vast amounts of text. Their ability to generate coherent "
    "text, answer questions and perform reasoning tasks has made them fundamental tools "
    "in the technology industry. However, they come with challenges such as hallucinations, "
    "bias and high computational cost."
)

result = summarize_with_eval.invoke(text)
print(f"Status: {result['status']}")
print(f"Iterations: {result['iterations']}")
print(f"Clarity: {result['final_evaluation']['clarity_score']}")
print(f"Completeness: {result['final_evaluation']['completeness_score']}")
print(f"Summary: {result['summary']}")
# Expected output:
# Status: approved
# Iterations: 1-2
# Clarity: 0.8+
# Completeness: 0.8+
# Summary: LLMs, based on the transformer architecture, have revolutionized...

Summary

In this capsule you learned 5 reusable patterns for the Functional API:

• Pattern 1: Tool execution loop — The ReAct loop implemented with while True and @task. The tools run in parallel with Futures. It's the base pattern for any agent
• Pattern 2: Multi-step reasoning — Decompose → reason → synthesize. Each step as a @task with automatic checkpointing. If it fails at step 3, steps 1 and 2 don't get re-run
• Pattern 3: Parallel task execution — Launch multiple @task without .result(), collect afterwards. The difference between sequential and parallel is where you put the .result()
• Pattern 4: Structured output — Pydantic + with_structured_output() inside a @task. Use .model_dump() so the output is serializable for checkpointing
• Pattern 5: Error handling with fallback — try/except + a for loop for retry with model escalation. Plain Python for resilience, no conditional edges

When these patterns aren't enough — complex branches, Send API, reusable sub-workflows, automatic visualization — it's time to migrate to the Graph API.

Next capsule: Mixing APIs — how to use a StateGraph inside an @entrypoint and vice versa, to get the best of both APIs in the same project.

Additional resources

1. Functional API Conceptual Guide — Core concepts of @entrypoint and @task
2. Workflows and Agents Patterns — Official patterns (parallelization, routing, evaluator-optimizer)
3. How to use the Functional API — Practical tutorial with complete examples
4. LangGraph Persistence — How the checkpointing that makes these patterns possible actually works
5. Pydantic with_structured_output — Structured output with LangChain and Pydantic
6. LangGraph Error Handling — Error handling and retry in LangGraph
7. Python Futures (concurrent.futures) — Futures reference in Python (the concept @task is analogous to)

Module 6 — LangChain & LangGraph: From Chains to Agents