@task: The Units That Make Up the Agent

Capsule overview

In the previous capsule you learned @entrypoint: the entry point of your functional workflow. But an @entrypoint that does everything in a single function doesn't take advantage of LangGraph's guarantees — checkpointing, progress streaming, parallel execution. For that you need to break your workflow into tasks.

@task defines independent units of work inside your workflow. Each task is:

• Checkpointable — if your workflow crashes after task A completes, you don't need to re-run it. LangGraph saved its result.
• Streamable — you can report progress to the user while the tasks run.
• Composable — you can run multiple tasks in parallel and combine their results.

Think of @task as the building blocks inside your @entrypoint. The entrypoint is the blueprint of the building; the tasks are the bricks.

There's one concept you need to internalize before writing any code: @task doesn't return the result directly — it returns a Future. If you don't get this, you'll spend hours debugging why your code "works" but produces strange objects instead of strings. This capsule will make it crystal clear.

Import

from langgraph.func import entrypoint, task

task lives in the same module as entrypoint. You import them together because you always use them together.

Basic usage: @task inside @entrypoint

Let's start with a complete example that shows the fundamental pattern:

from dotenv import load_dotenv
load_dotenv()

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

@task
def search_web(query: str) -> str:
    """Searches the web for information."""
    return f"Results for '{query}': LangChain is a framework for building applications with LLMs. It supports multiple providers and has orchestration tooling."

@task
def summarize(text: str) -> str:
    """Produces a summary of the text using an LLM."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Summarize this in one sentence: {text}")
    return response.content

@entrypoint()
def research(topic: str) -> str:
    search_result = search_web(topic).result()
    summary = summarize(search_result).result()
    return summary

result = research.invoke("LangChain")
print(result)
# Expected output: LangChain is a framework that lets you build applications
# with LLMs using multiple providers and orchestration tooling.

Step by step:

1. search_web is decorated with @task — it's an independent unit of work
2. summarize is also a @task — another independent unit
3. Inside the @entrypoint, you call each task and call .result() to get the value
4. The @entrypoint orchestrates: first search, then summarize

The obvious question: why .result()? Why can't I just use the value directly?

The Future concept: @task does NOT return the result

This is the most important concept in this capsule. Read it twice if you need to.

When you call a function decorated with @task, you don't get the function's result. You get a Future — an object that promises the result will be available, but doesn't have it yet.

The analogy: JavaScript Promises

If you know JavaScript, a Future is exactly like a Promise:

// JavaScript — Promise
const resultPromise = fetch("https://api.example.com/data");
// resultPromise is NOT the data — it's a Promise
// You need: const data = await resultPromise;

# Python/LangGraph — Future
result_future = search_web("LangChain")
# result_future is NOT the string — it's a Future
# You need: result = result_future.result()

Why Futures? Three concrete reasons

1. Checkpointing between tasks

If task A completes and returns its Future, LangGraph can save that result. If task B fails, you can re-run the workflow and LangGraph skips task A (it already has its saved result). Without Futures, LangGraph wouldn't have a cut point between tasks.

2. Parallel execution

When you call two tasks without immediately asking for .result(), LangGraph can run them in parallel:

@entrypoint()
def research(topic: str) -> str:
    future_a = search_web(topic)       # Launches task A
    future_b = search_news(topic)      # Launches task B (in parallel)
    
    result_a = future_a.result()       # Waits for result A
    result_b = future_b.result()       # Waits for result B
    
    return f"{result_a}\n{result_b}"

If @task returned the result directly, the second task wouldn't start until the first one finished. Futures let you launch both and wait afterwards.

3. Progress streaming

LangGraph can tell the user: "task search_web started", "task search_web completed", "task summarize started"... This is possible because each task is a discrete unit with a beginning and an end, thanks to Futures.

What happens without .result()

Let's see what happens when you forget to call .result():

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def greet(name: str) -> str:
    return f"Hello, {name}!"

@entrypoint()
def my_workflow(name: str) -> str:
    result = greet(name)         # No .result()
    return f"Answer: {result}"

output = my_workflow.invoke("Ana")
print(output)
# Output: Answer: <langgraph.types.Future object at 0x...>

Instead of "Answer: Hello, Ana!", you get a Future object coerced into a string. Your code doesn't raise an error — it just produces garbage. This is the quietest, most frustrating bug in the Functional API.

The fix:

@entrypoint()
def my_workflow(name: str) -> str:
    result = greet(name).result()  # ✅ With .result()
    return f"Answer: {result}"

output = my_workflow.invoke("Ana")
print(output)
# Expected output: Answer: Hello, Ana!

Golden rule: every time you call a @task, the next operation should be .result() (unless you're intentionally launching tasks in parallel).

A complete example: a research workflow with multiple tasks

An example that justifies using @task — each task can fail independently, is expensive (it calls APIs), and could run in parallel:

from dotenv import load_dotenv
load_dotenv()

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

@task
def decompose_query(topic: str) -> list[str]:
    """Breaks a topic down into research sub-questions."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Generate exactly 3 research sub-questions about: {topic}. "
        f"Return only the questions, one per line, unnumbered."
    )
    questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return questions[:3]

@task
def research_question(question: str) -> str:
    """Researches a single question using an LLM."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Answer this research question in 2-3 sentences: {question}"
    )
    return response.content

@task
def synthesize(findings: list[str]) -> str:
    """Synthesizes multiple findings into a coherent summary."""
    model = init_chat_model("openai:gpt-4.1-mini")
    combined = "\n\n".join(f"Finding {i+1}: {f}" for i, f in enumerate(findings))
    response = model.invoke(
        f"Synthesize these research findings into a coherent paragraph:\n\n{combined}"
    )
    return response.content

@entrypoint()
def research_agent(topic: str) -> str:
    questions = decompose_query(topic).result()
    
    findings = []
    for question in questions:
        finding = research_question(question).result()
        findings.append(finding)
    
    summary = synthesize(findings).result()
    return summary

result = research_agent.invoke("The impact of AI on education")
print(result)
# Expected output: A coherent paragraph synthesizing findings about
# the impact of AI on education, covering the 3 sub-questions.

Why does each function need @task here?

• decompose_query: calls an LLM (expensive, can fail on a rate limit)
• research_question: calls an LLM once per question (can fail independently)
• synthesize: calls an LLM (if it fails, we don't want to re-run the research)

If decompose_query and the 3 research_question calls complete but synthesize fails, LangGraph can re-run only synthesize thanks to per-task checkpointing.

Parallel execution with Futures

The previous example researches the questions in sequence. But because research_question is a @task that returns a Future, you can launch them all and wait afterwards:

from dotenv import load_dotenv
load_dotenv()

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

@task
def search_source(source: str, query: str) -> str:
    """Searches a specific source."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Pretend you are the source '{source}'. Answer briefly: {query}"
    )
    return f"[{source}] {response.content}"

@task
def merge_results(results: list[str]) -> str:
    """Combines results from multiple sources."""
    return "Combined results:\n" + "\n".join(f"  • {r}" for r in results)

@entrypoint()
def multi_source_search(query: str) -> str:
    sources = ["Wikipedia", "ArXiv", "StackOverflow"]
    
    futures = []
    for source in sources:
        future = search_source(source, query)  # Launch without .result()
        futures.append(future)
    
    results = [f.result() for f in futures]  # Wait for all of them
    
    merged = merge_results(results).result()
    return merged

output = multi_source_search.invoke("What is RAG?")
print(output)
# Expected output:
# Combined results:
#   • [Wikipedia] RAG (Retrieval-Augmented Generation) is a technique...
#   • [ArXiv] RAG combines document retrieval with generation...
#   • [StackOverflow] RAG is used to give the LLM relevant context...

The pattern is clear:

1. Launch — you call the tasks in a loop without .result()
2. Collect — you store the Futures in a list
3. Wait — you call .result() on each Future when you need the values

When to use @task vs regular functions

Not every function needs to be a task. Using @task unnecessarily adds overhead (serialization, checkpointing) with no benefit.

An example of when not to use @task:

from dotenv import load_dotenv
load_dotenv()

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

def format_as_markdown(text: str) -> str:
    """Regular function — simple, fast, can't fail."""
    return f"## Result\n\n{text}"

def validate_input(topic: str) -> str:
    """Regular function — trivial validation."""
    if len(topic.strip()) < 3:
        raise ValueError("The topic must be at least 3 characters long")
    return topic.strip()

@task
def generate_analysis(topic: str) -> str:
    """@task — calls an LLM, can fail, is expensive."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Briefly analyze this topic: {topic}")
    return response.content

@entrypoint()
def analyze(topic: str) -> str:
    clean_topic = validate_input(topic)       # Regular function — no .result()
    analysis = generate_analysis(clean_topic).result()  # @task — with .result()
    formatted = format_as_markdown(analysis)  # Regular function — no .result()
    return formatted

result = analyze.invoke("Artificial Intelligence")
print(result)
# Expected output:
# ## Result
#
# Artificial Intelligence is a field of computer science that seeks to
# create systems capable of performing tasks that normally require
# human intelligence...

Regular functions are called directly (no .result()). Only @tasks return Futures.

Tasks as checkpointable units

One of the main reasons to use @task is checkpointing. When you configure a checkpointer (you'll see this in detail in Module 8), LangGraph saves the result of every completed task.

The concept is simple:

Workflow: task_A → task_B → task_C

First run:
  ✅ task_A completes → result saved
  ✅ task_B completes → result saved
  ❌ task_C fails (API error)

Second run (with a checkpointer):
  ⏭️ task_A — result recovered from the checkpoint (not re-run)
  ⏭️ task_B — result recovered from the checkpoint (not re-run)
  ✅ task_C — re-run (now it works)

For checkpointing to work, you need two things:

1. Your units of work must be decorated with @task
2. You must configure a checkpointer when compiling (Module 8)

Without @task, LangGraph has no cut points to save progress. The whole @entrypoint is a single unit — if it fails anywhere, everything re-runs.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def step_a() -> str:
    print("  Running step_a...")
    return "result_a"

@task
def step_b(input: str) -> str:
    print("  Running step_b...")
    return f"result_b (based on {input})"

@task
def step_c(input: str) -> str:
    print("  Running step_c...")
    return f"result_c (based on {input})"

@entrypoint()
def pipeline(start: str) -> str:
    a = step_a().result()
    b = step_b(a).result()
    c = step_c(b).result()
    return c

result = pipeline.invoke("start")
print(result)
# Expected output:
#   Running step_a...
#   Running step_b...
#   Running step_c...
# result_c (based on result_b (based on result_a))

Each print shows you that the task ran. With a checkpointer configured (Module 8), on the second run you'd only see the prints of the tasks that need re-running.

@task restrictions

Tasks CANNOT be nested

A @task can't call another @task internally. Tasks are flat — they all live at the same level inside the @entrypoint:

from langgraph.func import entrypoint, task

@task
def outer_task(x: str) -> str:
    result = inner_task(x).result()  # ❌ ERROR: task inside a task
    return result

@task
def inner_task(x: str) -> str:
    return x.upper()

The fix is to call both tasks from the @entrypoint:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def process(x: str) -> str:
    return x.upper()

@task
def enrich(x: str) -> str:
    return f"[Enriched] {x}"

@entrypoint()
def workflow(text: str) -> str:
    processed = process(text).result()
    enriched = enrich(processed).result()
    return enriched

result = workflow.invoke("hello world")
print(result)
# Expected output: [Enriched] HELLO WORLD

Return values must be serializable

@task results are saved for checkpointing and streaming. That means they must be convertible to JSON:

from langgraph.func import task

@task
def good_task() -> dict:
    return {"status": "ok", "items": [1, 2, 3]}  # ✅ Serializable

@task
def bad_task():
    import sqlite3
    return sqlite3.connect(":memory:")  # ❌ Not serializable

Tasks need a parent @entrypoint

You can't run a @task outside an @entrypoint. The entrypoint is the context that enables checkpointing and streaming:

from langgraph.func import task

@task
def my_task(x: str) -> str:
    return x.upper()

# This does NOT work outside an @entrypoint:
# result = my_task("hello").result()  # ❌ Error: no entrypoint context

Pattern: a task with manual retry

Until you have Module 7's retry system, you can implement basic retry inside a task:

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 call_llm_with_retry(prompt: str, max_retries: int = 3) -> str:
    """Calls the LLM with manual retries."""
    model = init_chat_model("openai:gpt-4.1-mini")
    for attempt in range(max_retries):
        try:
            response = model.invoke(prompt)
            return response.content
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"  Attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
                time.sleep(wait)
            else:
                raise RuntimeError(f"Failed after {max_retries} attempts: {e}")

@entrypoint()
def robust_workflow(topic: str) -> str:
    result = call_llm_with_retry(f"Briefly explain: {topic}").result()
    return result

output = robust_workflow.invoke("Futures in Python")
print(output)
# Expected output: Futures in Python are objects that represent
# the result of an asynchronous operation that hasn't completed yet...

The retry lives inside the task because the task is the unit that can fail. The @entrypoint doesn't need to know the retry details.

Troubleshooting

Problem 1: "My code returns a Future object instead of the value"

Symptom: Instead of a string like "Hello!", you get something like <langgraph.types.Future object at 0x...>. Cause: You forgot to call .result() after invoking a @task. Fix:

# ❌ Without .result() — you get a Future
result = my_task("input")

# ✅ With .result() — you get the real value
result = my_task("input").result()

Problem 2: "Error: task called outside an entrypoint"

Symptom: An error when trying to run a @task directly. Cause: Tasks can only run inside an @entrypoint. Fix: Make sure the task call is inside a function decorated with @entrypoint():

# ❌ Loose task
output = my_task("data").result()

# ✅ Task inside an entrypoint
@entrypoint()
def my_workflow(data: str) -> str:
    return my_task(data).result()

output = my_workflow.invoke("data")

Problem 3: "Serialization error when returning from a task"

Symptom: An error saying the return value can't be serialized. Cause: The task returns a non-serializable object (a DB connection, a function, a generator). Fix: Return primitive types, lists, dictionaries, or Pydantic models:

# ❌ Returns a non-serializable object
@task
def bad() -> object:
    return open("file.txt")

# ✅ Returns serializable data
@task
def good() -> dict:
    with open("file.txt") as f:
        return {"content": f.read()}

Problem 4: "I tried nesting tasks and got an error"

Symptom: An error when calling a @task from inside another @task. Cause: Tasks are flat — they can't be nested. Fix: Move both tasks up to the @entrypoint level and orchestrate from there:

# ❌ Task inside a task
@task
def outer(x):
    return inner(x).result()

# ✅ Both at the same level, orchestrated by the entrypoint
@entrypoint()
def workflow(x):
    a = task_a(x).result()
    b = task_b(a).result()
    return b

Problem 5: "My tasks run in sequence even though I want them parallel"

Symptom: Tasks run one after another, not in parallel. Cause: You're calling .result() immediately after each task. Fix: Launch all the tasks first, then call .result():

# ❌ Sequential — each .result() blocks before the next launch
r1 = task_a("x").result()
r2 = task_b("y").result()

# ✅ Parallel — launch everything, then wait
f1 = task_a("x")
f2 = task_b("y")
r1 = f1.result()
r2 = f2.result()

Exercises

Exercise 1: Your first @task (Easy)

Create a @task called translate that takes a text and returns a "simulated translation" (it prepends the [EN] prefix to the text). Use it inside an @entrypoint that translates a greeting. Don't forget .result().

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def translate(text: str) -> str:
    """Simulates a translation by adding a prefix."""
    return f"[EN] {text}"

@entrypoint()
def translate_workflow(text: str) -> str:
    translated = translate(text).result()
    return translated

result = translate_workflow.invoke("Hi, how are you?")
print(result)
# Expected output: [EN] Hi, how are you?

Exercise 2: Spot the Future bug (Easy)

The following code has a subtle bug. Find it and fix it:

from langgraph.func import entrypoint, task

@task
def analyze(text: str) -> dict:
    word_count = len(text.split())
    return {"text": text, "word_count": word_count}

@entrypoint()
def analyze_workflow(text: str) -> str:
    analysis = analyze(text)
    return f"Words: {analysis['word_count']}"

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def analyze(text: str) -> dict:
    word_count = len(text.split())
    return {"text": text, "word_count": word_count}

@entrypoint()
def analyze_workflow(text: str) -> str:
    analysis = analyze(text).result()  # ✅ .result() was missing
    return f"Words: {analysis['word_count']}"

result = analyze_workflow.invoke("LangGraph is a powerful framework")
print(result)
# Expected output: Words: 5

Exercise 3: A pipeline of 3 sequential tasks (Medium)

Create a workflow with 3 tasks: extract_keywords (pulls the 3 longest words from a text), format_keywords (uppercases them and joins them with commas), and generate_report (builds a report with the keywords). Orchestrate everything from an @entrypoint.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def extract_keywords(text: str) -> list[str]:
    """Extracts the 3 longest words as keywords."""
    words = text.split()
    sorted_words = sorted(words, key=len, reverse=True)
    return sorted_words[:3]

@task
def format_keywords(keywords: list[str]) -> str:
    """Formats keywords in uppercase, comma-separated."""
    return ", ".join(kw.upper() for kw in keywords)

@task
def generate_report(original: str, formatted_kw: str) -> str:
    """Generates a report with the text and its keywords."""
    return f"Analyzed text: '{original}'\nDetected keywords: {formatted_kw}"

@entrypoint()
def keyword_pipeline(text: str) -> str:
    keywords = extract_keywords(text).result()
    formatted = format_keywords(keywords).result()
    report = generate_report(text, formatted).result()
    return report

result = keyword_pipeline.invoke("LangGraph lets you build intelligent applications")
print(result)
# Expected output:
# Analyzed text: 'LangGraph lets you build intelligent applications'
# Detected keywords: APPLICATIONS, INTELLIGENT, LANGGRAPH

Exercise 4: Parallel search across 3 sources (Medium)

Create 3 tasks that simulate searching different sources (Wikipedia, ArXiv, GitHub). Launch all 3 in parallel inside the @entrypoint, collect the results, and combine them into a single string.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def search_wikipedia(query: str) -> str:
    """Simulates a Wikipedia search."""
    return f"[Wikipedia] '{query}': Encyclopedic article with a general definition and historical context."

@task
def search_arxiv(query: str) -> str:
    """Simulates an ArXiv search."""
    return f"[ArXiv] '{query}': Recent paper on technical advances and benchmarks."

@task
def search_github(query: str) -> str:
    """Simulates a GitHub search."""
    return f"[GitHub] '{query}': Repository with a reference implementation and examples."

@entrypoint()
def parallel_search(query: str) -> str:
    future_wiki = search_wikipedia(query)
    future_arxiv = search_arxiv(query)
    future_github = search_github(query)
    
    results = [
        future_wiki.result(),
        future_arxiv.result(),
        future_github.result(),
    ]
    
    return "Search results:\n" + "\n".join(f"  • {r}" for r in results)

output = parallel_search.invoke("Retrieval Augmented Generation")
print(output)
# Expected output:
# Search results:
#   • [Wikipedia] 'Retrieval Augmented Generation': Encyclopedic article...
#   • [ArXiv] 'Retrieval Augmented Generation': Recent paper...
#   • [GitHub] 'Retrieval Augmented Generation': Repository with a reference implementation...

Exercise 5: @task vs regular function — refactor (Medium)

The following code uses @task for everything, including trivial operations. Refactor it: keep @task only where it's justified and turn the rest into regular functions.

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

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

@task
def add_timestamp(text: str) -> str:
    from datetime import datetime
    return f"[{datetime.now().isoformat()}] {text}"

from dotenv import load_dotenv
load_dotenv()

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

def clean_input(text: str) -> str:
    """Regular function — trivial operation, can't fail, not expensive."""
    return text.strip().lower()

@task
def call_model(prompt: str) -> str:
    """@task — calls an external API, expensive, can fail."""
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.invoke(prompt).content

def add_timestamp(text: str) -> str:
    """Regular function — trivial operation, instantaneous."""
    return f"[{datetime.now().isoformat()}] {text}"

@entrypoint()
def process(text: str) -> str:
    cleaned = clean_input(text)                   # Regular — no .result()
    response = call_model(cleaned).result()       # @task — with .result()
    timestamped = add_timestamp(response)         # Regular — no .result()
    return timestamped

result = process.invoke("  What is LangGraph?  ")
print(result)
# Expected output: [2026-03-08T...] LangGraph is a framework for building
# applications with LLMs using state graphs...

Exercise 6: A mini research agent with decompose + parallel search + synthesize (Advanced)

Build a complete workflow: one task breaks a topic into 2 questions, two tasks research each question in parallel (simulated), and a final task synthesizes the findings. Use the parallel Futures pattern.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def decompose(topic: str) -> list[str]:
    """Breaks a topic into 2 research questions."""
    return [
        f"What is the definition and origin of {topic}?",
        f"What are the practical applications of {topic}?",
    ]

@task
def investigate(question: str) -> str:
    """Researches a question (simulated)."""
    if "definition" in question:
        return f"Finding: {question.split('of ')[-1].rstrip('?')} refers to a field of technology with roots in the 1950s."
    return f"Finding: It is applied in medicine, finance, education, and autonomous transportation."

@task
def synthesize(findings: list[str]) -> str:
    """Synthesizes findings into a final summary."""
    summary = "Research summary:\n"
    for i, finding in enumerate(findings, 1):
        summary += f"  {i}. {finding}\n"
    summary += "Conclusion: A topic with broad impact across multiple industries."
    return summary

@entrypoint()
def mini_research(topic: str) -> str:
    questions = decompose(topic).result()
    
    futures = [investigate(q) for q in questions]
    findings = [f.result() for f in futures]
    
    report = synthesize(findings).result()
    return report

result = mini_research.invoke("Artificial Intelligence")
print(result)
# Expected output:
# Research summary:
#   1. Finding: Artificial Intelligence refers to a field of technology with roots in the 1950s.
#   2. Finding: It is applied in medicine, finance, education, and autonomous transportation.
# Conclusion: A topic with broad impact across multiple industries.

Summary

In this capsule you learned:

• @task defines independent units of work inside an @entrypoint — each task is checkpointable, streamable, and composable
• @task returns a Future, not the result directly — you must call .result() to get the real value
• A Future is like a JavaScript Promise: the result is on its way but isn't ready yet
• Futures enable: checkpointing (saving progress between tasks), parallel execution (launching several tasks and waiting afterwards), and streaming (reporting progress per task)
• Without .result() you get a Future object coerced into a string — the quietest bug in the Functional API
• Use @task when the operation can fail (APIs), is expensive (LLMs), or can run in parallel. Use regular functions for trivial operations
• Tasks are flat: they can't be nested (a task inside a task)
• Return values must be serializable (strings, dicts, lists — not connections or functions)
• For parallel execution: launch the tasks without .result(), store the Futures in a list, and call .result() at the end

Next capsule: Native Control Flow — you'll learn to use Python's while, if/else, for, and try/except to steer your workflow's execution, without edges or conditional edges.

Additional resources

1. LangGraph Functional API — Conceptual Guide — Official docs for @entrypoint and @task
2. LangGraph Functional API — How-To Guide — Step-by-step tutorial with the Functional API
3. Python concurrent.futures — Future objects — Reference for Futures in standard Python (a similar concept)
4. JavaScript Promises — MDN — Promise reference for the analogy
5. LangGraph Checkpointing — How checkpointing works with tasks (a preview of Module 8)
6. LangGraph Streaming — Progress streaming per task
7. init_chat_model — LangChain — Reference for init_chat_model, used in the examples

Module 6 — LangChain & LangGraph: From Chains to Agents