Graph API vs Functional API: A Deep Comparison

Capsule overview

You already know both LangGraph APIs. In Module 5 you built workflows with StateGraph, nodes and edges. In the previous capsules of this module you learned @entrypoint, @task, .result(), and control flow with plain Python. Now comes the inevitable question: when do I use each one?

The answer isn't "one is better than the other." They're two ways of expressing the same intent — building workflows with durability, checkpointing and streaming. The Graph API gives you an explicit graph with nodes and edges that you can visualize. The Functional API gives you Python functions with decorators that read like sequential code. Both use the same runtime underneath. The right decision depends on the problem, not on a blanket preference.

In this capsule you're going to see a detailed comparison table, three problems solved side-by-side with both APIs, a concrete decision framework, and the migration path between them.

Detailed comparison table

The table is a starting point, not a verdict. Let's see in code how these differences actually show up.

Side-by-side #1: Chatbot with tools

The most common pattern in agents: a model that decides whether to call tools, runs the tools, and goes back to the model until it has a final answer. The ReAct loop.

Graph API version

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 current weather for a city."""
    return f"Sunny, 22°C in {city}"

@tool
def calculator(expression: str) -> str:
    """Evaluates a math expression."""
    return str(eval(expression))

tools_list = [get_weather, calculator]
tool_map = {t.name: t for t in tools_list}

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

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

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

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}

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

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

result = app.invoke({"messages": [HumanMessage(content="What's the weather in Madrid, and how much is 15 * 7?")]})
print(result["messages"][-1].content)
# Expected output: The weather in Madrid is sunny with 22°C. And 15 × 7 = 105.

Functional API version

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, BaseMessage
from langchain_core.tools import tool

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

@tool
def calculator(expression: str) -> str:
    """Evaluates a math expression."""
    return str(eval(expression))

tools_list = [get_weather, calculator]
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 call_tool(tool_call):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def chatbot(messages: list[BaseMessage]):
    llm_response = call_model(messages).result()

    while llm_response.tool_calls:
        tool_futures = [call_tool(tc) for tc in llm_response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]
        messages = add_messages(messages, [llm_response, *tool_results])
        llm_response = call_model(messages).result()

    return llm_response.content

result = chatbot.invoke([HumanMessage(content="What's the weather in Madrid, and how much is 15 * 7?")])
print(result)
# Expected output: The weather in Madrid is sunny with 22°C. And 15 × 7 = 105.

Analysis

Both versions do exactly the same thing. The Graph API gives you a visual diagram of the loop. The Functional API runs the tools in parallel with Futures and reads like a standard while loop. Neither is objectively better — it depends on whether you value visualization or code readability more.

Side-by-side #2: Research agent with routing

An agent that classifies the user's question and routes it to a different specialist depending on the type.

Graph API version

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, SystemMessage
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 ["code", "program", "bug", "function"]):
        return {"intent": "code"}
    elif any(w in text for w in ["data", "statistic", "number", "analyze"]):
        return {"intent": "data"}
    return {"intent": "general"}

def route_intent(state: State) -> str:
    return state["intent"]

def make_specialist(system_prompt: str):
    def specialist(state: State) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke([SystemMessage(content=system_prompt)] + state["messages"])]}
    return specialist

graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("code_expert", make_specialist("You are a code expert. Answer in English."))
graph.add_node("data_analyst", make_specialist("You are a data analyst. Answer in English."))
graph.add_node("generalist", make_specialist("You are a general assistant. Answer in English."))

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
    "code": "code_expert", "data": "data_analyst", "general": "generalist"
})
for node in ["code_expert", "data_analyst", "generalist"]:
    graph.add_edge(node, END)

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

result = app.invoke({
    "messages": [HumanMessage(content="How do I write a recursive function in Python?")],
    "intent": ""
})
print(f"Intent: {result['intent']}")
print(result["messages"][-1].content[:100])
# Expected output:
# Intent: code
# A recursive function is one that calls itself. Here's an example...

Functional API version

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

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

def classify_intent(text: str) -> str:
    text_lower = text.lower()
    if any(w in text_lower for w in ["code", "program", "bug", "function"]):
        return "code"
    elif any(w in text_lower for w in ["data", "statistic", "number", "analyze"]):
        return "data"
    return "general"

@task
def ask_specialist(question: str, system_prompt: str) -> str:
    response = model.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question),
    ])
    return response.content

SPECIALISTS = {
    "code": "You are a code expert. Answer in English.",
    "data": "You are a data analyst. Answer in English.",
    "general": "You are a general assistant. Answer in English.",
}

@entrypoint()
def research_agent(question: str) -> dict:
    intent = classify_intent(question)
    answer = ask_specialist(question, SPECIALISTS[intent]).result()
    return {"intent": intent, "answer": answer}

result = research_agent.invoke("How do I write a recursive function in Python?")
print(f"Intent: {result['intent']}")
print(result["answer"][:100])
# Expected output:
# Intent: code
# A recursive function is one that calls itself. Here's an example...

Analysis

With 3 routes, both APIs work well. The Graph API shows you the diagram with the 3 routes leaving the classifier — useful for presenting the flow to a team. The Functional API condenses everything into a dict lookup plus a @task call.

Side-by-side #3: Multi-step pipeline with evaluation

A pipeline that generates content, evaluates it, and regenerates it if it doesn't meet the quality bar. It includes a conditional improvement loop.

Graph API version

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class State(TypedDict):
    topic: str
    draft: str
    feedback: str
    quality: str
    iterations: int

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

def generate(state: State) -> dict:
    prompt = f"Write a paragraph about: {state['topic']}"
    if state.get("feedback"):
        prompt += f"\nImprove it based on this feedback: {state['feedback']}"
    response = model.invoke(prompt)
    return {"draft": response.content, "iterations": state.get("iterations", 0) + 1}

def evaluate(state: State) -> dict:
    response = model.invoke(
        f"Evaluate this text. Answer ONLY 'good' or 'needs_improvement' followed by brief feedback.\n\n{state['draft']}"
    )
    text = response.content.strip().lower()
    if "good" in text[:10]:
        return {"quality": "good", "feedback": ""}
    return {"quality": "needs_improvement", "feedback": response.content}

def route_quality(state: State) -> str:
    if state["quality"] == "good" or state.get("iterations", 0) >= 3:
        return "done"
    return "regenerate"

graph = StateGraph(State)
graph.add_node("generate", generate)
graph.add_node("evaluate", evaluate)
graph.add_edge(START, "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", route_quality, {
    "done": END, "regenerate": "generate"
})

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

result = app.invoke({"topic": "what a transformer is", "draft": "", "feedback": "", "quality": "", "iterations": 0})
print(f"Iterations: {result['iterations']}")
print(f"Quality: {result['quality']}")
print(result["draft"][:120])
# Expected output:
# Iterations: 1-3 (depends on the evaluation)
# Quality: good
# A transformer is a neural network architecture proposed in 2017...

Functional API version

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 generate_draft(topic: str, feedback: str = "") -> str:
    prompt = f"Write a paragraph about: {topic}"
    if feedback:
        prompt += f"\nImprove it based on this feedback: {feedback}"
    return model.invoke(prompt).content

@task
def evaluate_draft(draft: str) -> dict:
    response = model.invoke(
        f"Evaluate this text. Answer ONLY 'good' or 'needs_improvement' followed by brief feedback.\n\n{draft}"
    )
    text = response.content.strip().lower()
    if "good" in text[:10]:
        return {"quality": "good", "feedback": ""}
    return {"quality": "needs_improvement", "feedback": response.content}

@entrypoint()
def content_pipeline(topic: str) -> dict:
    feedback = ""
    iterations = 0
    max_iterations = 3

    while iterations < max_iterations:
        draft = generate_draft(topic, feedback).result()
        evaluation = evaluate_draft(draft).result()
        iterations += 1

        if evaluation["quality"] == "good":
            break
        feedback = evaluation["feedback"]

    return {"draft": draft, "iterations": iterations, "quality": evaluation["quality"]}

result = content_pipeline.invoke("what a transformer is")
print(f"Iterations: {result['iterations']}")
print(f"Quality: {result['quality']}")
print(result["draft"][:120])
# Expected output:
# Iterations: 1-3 (depends on the evaluation)
# Quality: good
# A transformer is a neural network architecture proposed in 2017...

Analysis

This example is where the difference is felt most. The Graph API shows a diagram with the circular evaluate → generate arrow that communicates the pattern visually. The Functional API expresses the same loop as a while with a break — more familiar to any Python developer, but invisible to anyone looking at the system from the outside.

Decision framework

Don't memorize rules — internalize criteria. When you start a project, ask yourself these questions:

Use the Graph API when:

• ✅ Your workflow has >5 nodes with complex routing between them
• ✅ You need to visualize the flow for debugging, documentation or communication with the team
• ✅ Multiple teams work on different nodes of the same workflow
• ✅ The topology is non-trivial: multiple branches, merge points, sub-workflows
• ✅ You need the Send API to create workers dynamically

Use the Functional API when:

• ✅ Your workflow is sequential with simple branching (if/else)
• ✅ You want a quick prototype that works in minutes
• ✅ You'd rather the code read like ordinary Python
• ✅ Your developers come from Python and don't know graph concepts
• ✅ The flow is naturally expressed as a loop with conditions

Both work when:

• ✅ Moderate complexity (3-5 steps with one branch)
• ✅ Team preference
• ✅ The flow might grow but today it's simple

The "can I sketch it easily?" rule

If you can draw your workflow on a napkin in 10 seconds and it looks like a line with a couple of branches, the Functional API is probably more direct. If it takes you 30 seconds and the drawing has decision diamonds, merges, and crossing arrows, the Graph API is going to save you headaches.

Migration path

The good news: this isn't a permanent decision. Both APIs share the same runtime, so migrating is refactoring, not rewriting.

From Functional API to Graph API

The natural progression:

1. You start with @entrypoint + @task
   → It works, the code is clean

2. You add more branches and loops
   → The if/elif/else grows, but it's manageable

3. You hit an inflection point:
   → "I need to visualize this flow to debug it"
   → "Another team needs to understand the topology"
   → "I have 8 tasks with complex routing between them"

4. You migrate to StateGraph
   → Each @task becomes a node
   → Each if/else becomes a conditional edge
   → The while loop becomes a circular edge

Signs you need to migrate

• ❌ Your @entrypoint has more than 40 lines of control logic
• ❌ You have nested if/elif/else more than 3 levels deep
• ❌ You need another team to understand the flow without reading the code
• ❌ Debugging requires putting prints in 10 different places

Signs you do NOT need to migrate

• ✅ Your flow is a simple loop with 2-3 tasks
• ✅ The code makes sense in a single linear read
• ✅ You're the only one working on this workflow
• ✅ You don't need to explain the flow visually to anyone

Can you mix them?

Yes. Both APIs share the same runtime, so you can use a compiled StateGraph inside an @entrypoint, or call a functional workflow from a node of a graph. This is extremely useful when parts of your system are better expressed as a graph and others as sequential functions.

This pattern is covered in detail in capsule 07 of this module. For now, what matters is knowing that it's not an all-or-nothing decision — you can combine them in the same project.

Troubleshooting

Problem 1: "I chose the Functional API but now I have 6 levels of if/else"

Symptom: Your @entrypoint grew to 80+ lines with complex branching and is hard to follow.

Fix: It's a sign you need to migrate to the Graph API. Each branch of your if/else is a candidate to become an independent node. Each condition is a conditional edge. The migration is direct:

# Before (Functional): nested if/else
if intent == "code":
    if language == "python":
        result = python_expert(query).result()
    else:
        result = general_code(query).result()
elif intent == "data":
    result = data_analyst(query).result()

# After (Graph): explicit conditional edges
graph.add_conditional_edges("classify", route_intent, {
    "python_code": "python_expert",
    "general_code": "general_code",
    "data": "data_analyst",
})

Problem 2: "I chose the Graph API but my graph is a straight line"

Symptom: Your StateGraph has 4 nodes all connected with fixed edges: START → A → B → C → END. There are no conditional edges.

Fix: A linear graph with no branching is a sign of over-engineering. Migrate it to the Functional API, where it's expressed as 3 sequential calls:

@entrypoint()
def pipeline(input_data: str) -> str:
    a = task_a(input_data).result()
    b = task_b(a).result()
    return task_c(b).result()

Problem 3: "I don't know if my flow is 'complex enough' for the Graph API"

Symptom: Analysis paralysis. Every project starts with 20 minutes deciding which API to use.

Fix: Always start with the Functional API. If in the first 30 minutes of development the control flow starts feeling tangled, migrate. The decision doesn't have to be right from the start — the migration is a refactor, not a rewrite.

Problem 4: "I need visualization but I prefer the Functional API"

Symptom: You want a diagram of the flow for documentation, but your workflow is sequential and the Functional API is more natural.

Fix: You can document the flow with a hand-written Mermaid diagram or use external tools. The automatic visualization from draw_mermaid_png() is convenient, but it isn't the only way to document a workflow. If visualization is the only reason for the Graph API, it probably doesn't justify the migration.

Exercises

Exercise 1: Which API would you pick? (Easy)

For each scenario, decide whether you'd use the Graph API or the Functional API. Justify your answer.

A) A chatbot that answers support questions using 2 tools (search the FAQ, create a ticket).

B) A system that receives documents, classifies them into 5 categories, and runs a different pipeline for each category. Each pipeline has 3-4 steps.

C) An agent that translates text, evaluates the quality of the translation, and re-translates if the quality is low.

D) A workflow that searches 4 sources in parallel and synthesizes the results.

Exercise 2: Convert Graph API to Functional API (Easy)

Convert this Graph API workflow to the Functional API:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

class State(TypedDict):
    text: str
    cleaned: str
    summary: str

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

def clean_text(state: State) -> dict:
    cleaned = " ".join(state["text"].split())
    return {"cleaned": cleaned}

def summarize(state: State) -> dict:
    response = model.invoke(f"Summarize in one sentence: {state['cleaned']}")
    return {"summary": response.content}

graph = StateGraph(State)
graph.add_node("clean", clean_text)
graph.add_node("summarize", summarize)
graph.add_edge(START, "clean")
graph.add_edge("clean", "summarize")
graph.add_edge("summarize", END)

app = graph.compile()
result = app.invoke({"text": "  Python   is   a   very   popular   programming   language  "})
print(result["summary"])

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 clean_text(text: str) -> str:
    return " ".join(text.split())

@task
def summarize(text: str) -> str:
    return model.invoke(f"Summarize in one sentence: {text}").content

@entrypoint()
def text_pipeline(text: str) -> str:
    cleaned = clean_text(text).result()
    summary = summarize(cleaned).result()
    return summary

result = text_pipeline.invoke("  Python   is   a   very   popular   programming   language  ")
print(result)
# Expected output: Python is a very popular programming language.

Exercise 3: Convert Functional API to Graph API (Medium)

Convert this functional workflow to the Graph API. Justify why the migration makes sense.

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 classify(text: str) -> str:
    text_lower = text.lower()
    if any(w in text_lower for w in ["urgent", "error", "down"]):
        return "urgent"
    elif any(w in text_lower for w in ["invoice", "charge", "payment"]):
        return "billing"
    return "general"

@task
def handle_urgent(text: str) -> str:
    return model.invoke(f"URGENT. Answer fast and direct: {text}").content

@task
def handle_billing(text: str) -> str:
    return model.invoke(f"Billing question. Be precise: {text}").content

@task
def handle_general(text: str) -> str:
    return model.invoke(f"General question. Be friendly: {text}").content

@entrypoint()
def support_agent(text: str) -> dict:
    category = classify(text).result()
    if category == "urgent":
        response = handle_urgent(text).result()
    elif category == "billing":
        response = handle_billing(text).result()
    else:
        response = handle_general(text).result()
    return {"category": category, "response": response}

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class SupportState(TypedDict):
    text: str
    category: str
    response: str

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

def classify(state: SupportState) -> dict:
    text_lower = state["text"].lower()
    if any(w in text_lower for w in ["urgent", "error", "down"]):
        return {"category": "urgent"}
    elif any(w in text_lower for w in ["invoice", "charge", "payment"]):
        return {"category": "billing"}
    return {"category": "general"}

def route_category(state: SupportState) -> str:
    return state["category"]

def handle_urgent(state: SupportState) -> dict:
    return {"response": model.invoke(f"URGENT. Answer fast and direct: {state['text']}").content}

def handle_billing(state: SupportState) -> dict:
    return {"response": model.invoke(f"Billing question. Be precise: {state['text']}").content}

def handle_general(state: SupportState) -> dict:
    return {"response": model.invoke(f"General question. Be friendly: {state['text']}").content}

graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("handle_urgent", handle_urgent)
graph.add_node("handle_billing", handle_billing)
graph.add_node("handle_general", handle_general)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_category, {
    "urgent": "handle_urgent", "billing": "handle_billing", "general": "handle_general"
})
for node in ["handle_urgent", "handle_billing", "handle_general"]:
    graph.add_edge(node, END)

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

result = app.invoke({"text": "My server is down, I need urgent help"})
print(f"Category: {result['category']}")
print(result["response"][:80])
# Expected output:
# Category: urgent
# I understand the urgency. To diagnose the server problem...

Exercise 4: Implement the same problem in both APIs (Medium)

Implement a workflow that takes a topic, generates 3 questions about that topic in parallel (using an LLM), and combines them into a quiz. Implement it in both APIs and compare.

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 generate_question(topic: str, difficulty: str) -> str:
    return model.invoke(
        f"Generate ONE multiple-choice question about '{topic}' with {difficulty} difficulty. "
        f"Include 4 options (A-D) and mark the correct one."
    ).content

@task
def format_quiz(questions: list) -> str:
    quiz = "# Quiz\n\n"
    for i, q in enumerate(questions, 1):
        quiz += f"## Question {i}\n{q}\n\n"
    return quiz

@entrypoint()
def quiz_generator(topic: str) -> str:
    easy_fut = generate_question(topic, "easy")
    medium_fut = generate_question(topic, "medium")
    hard_fut = generate_question(topic, "hard")

    questions = [easy_fut.result(), medium_fut.result(), hard_fut.result()]
    return format_quiz(questions).result()

result = quiz_generator.invoke("Python decorators")
print(result[:200])
# Expected output:
# # Quiz
#
# ## Question 1
# What is a decorator in Python?
# A) A special class...

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

class QuizState(TypedDict):
    topic: str
    questions: Annotated[list[str], operator.add]
    quiz: str

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

def make_question_node(difficulty: str):
    def node(state: QuizState) -> dict:
        response = model.invoke(
            f"Generate ONE multiple-choice question about '{state['topic']}' with {difficulty} difficulty. "
            f"Include 4 options (A-D) and mark the correct one."
        )
        return {"questions": [response.content]}
    return node

def format_quiz(state: QuizState) -> dict:
    quiz = "# Quiz\n\n"
    for i, q in enumerate(state["questions"], 1):
        quiz += f"## Question {i}\n{q}\n\n"
    return {"quiz": quiz}

graph = StateGraph(QuizState)
graph.add_node("easy", make_question_node("easy"))
graph.add_node("medium", make_question_node("medium"))
graph.add_node("hard", make_question_node("hard"))
graph.add_node("format", format_quiz)

graph.add_edge(START, "easy")
graph.add_edge(START, "medium")
graph.add_edge(START, "hard")
graph.add_edge("easy", "format")
graph.add_edge("medium", "format")
graph.add_edge("hard", "format")
graph.add_edge("format", END)

app = graph.compile()
result = app.invoke({"topic": "Python decorators", "questions": [], "quiz": ""})
print(result["quiz"][:200])
# Expected output:
# # Quiz
#
# ## Question 1
# What is a decorator in Python?
# A) A special class...

Exercise 5: Automated decision framework (Advanced)

Write a function recommend_api(requirements: dict) -> str that takes a dictionary of requirements and returns "graph_api", "functional_api", or "either". The requirements are:

• num_nodes: estimated number of nodes/tasks
• has_complex_routing: bool (more than 2 conditional destinations)
• needs_visualization: bool
• is_sequential: bool (mostly linear flow)
• team_size: int (people working on the workflow)

Implement the decision logic and test it with at least 4 scenarios.

def recommend_api(requirements: dict) -> str:
    num_nodes = requirements.get("num_nodes", 3)
    has_complex_routing = requirements.get("has_complex_routing", False)
    needs_visualization = requirements.get("needs_visualization", False)
    is_sequential = requirements.get("is_sequential", True)
    team_size = requirements.get("team_size", 1)

    graph_signals = sum([
        num_nodes > 5,
        has_complex_routing,
        needs_visualization and num_nodes > 3,
        team_size > 2,
    ])

    functional_signals = sum([
        is_sequential,
        num_nodes <= 4,
        team_size <= 1,
        not has_complex_routing,
    ])

    if graph_signals >= 3:
        return "graph_api"
    if functional_signals >= 3 and graph_signals == 0:
        return "functional_api"
    return "either"


# Test 1: Simple pipeline
r1 = recommend_api({
    "num_nodes": 3, "has_complex_routing": False,
    "needs_visualization": False, "is_sequential": True, "team_size": 1
})
print(f"Simple pipeline: {r1}")
# Expected output: functional_api

# Test 2: Multi-route system with a team
r2 = recommend_api({
    "num_nodes": 8, "has_complex_routing": True,
    "needs_visualization": True, "is_sequential": False, "team_size": 4
})
print(f"Multi-route system: {r2}")
# Expected output: graph_api

# Test 3: Evaluate-improve loop
r3 = recommend_api({
    "num_nodes": 3, "has_complex_routing": False,
    "needs_visualization": False, "is_sequential": True, "team_size": 1
})
print(f"Evaluate-improve loop: {r3}")
# Expected output: functional_api

# Test 4: Moderate complexity
r4 = recommend_api({
    "num_nodes": 5, "has_complex_routing": True,
    "needs_visualization": True, "is_sequential": False, "team_size": 2
})
print(f"Moderate complexity: {r4}")
# Expected output: either

Exercise 6: Refactoring challenge (Advanced)

You have an @entrypoint that got out of hand. Identify the problems and decide: do you refactor within the Functional API, or migrate to the Graph API?

@entrypoint()
def complex_agent(input_data: dict) -> dict:
    text = input_data["text"]
    mode = input_data["mode"]

    if mode == "research":
        sources = search_web(text).result()
        if len(sources) > 3:
            summaries = []
            for s in sources[:5]:
                summaries.append(summarize_source(s).result())
            combined = merge_summaries(summaries).result()
        else:
            combined = summarize_source(sources[0]).result()
        quality = evaluate_quality(combined).result()
        if quality["score"] < 0.7:
            combined = improve_text(combined, quality["feedback"]).result()
        return {"result": combined, "mode": "research"}
    elif mode == "code":
        spec = generate_spec(text).result()
        code = generate_code(spec).result()
        tests = generate_tests(code).result()
        test_result = run_tests(code, tests).result()
        if not test_result["passed"]:
            code = fix_code(code, test_result["errors"]).result()
        return {"result": code, "mode": "code"}
    elif mode == "translate":
        translation = translate(text).result()
        back_translation = translate(translation).result()
        if similarity(text, back_translation) < 0.8:
            translation = translate(text).result()
        return {"result": translation, "mode": "translate"}
    return {"result": "Unsupported mode", "mode": mode}

@entrypoint()
def complex_agent(input_data: dict) -> dict:
    mode = input_data["mode"]
    text = input_data["text"]

    handlers = {
        "research": research_workflow,
        "code": code_workflow,
        "translate": translate_workflow,
    }

    handler = handlers.get(mode)
    if handler is None:
        return {"result": "Unsupported mode", "mode": mode}

    result = handler(text).result()
    return {"result": result, "mode": mode}

Summary

In this capsule you learned:

• The Graph API and the Functional API share the same runtime — the difference is how you express the workflow, not how it runs
• The Graph API gives you explicit structure: nodes, edges, conditional edges, visualization with draw_mermaid_png()
• The Functional API gives you idiomatic Python: while, if/else, for, Futures for parallelism
• You saw 3 problems solved side-by-side: chatbot with tools, agent with routing, pipeline with evaluation
• There's no "better" API — there are criteria: topology complexity, need for visualization, team size, nature of the flow
• Decision framework: >5 nodes + complex routing + large team → Graph API. Sequential flow + simple branching + solo developer → Functional API
• The migration path goes from Functional to Graph when complexity justifies it — it's a refactor, not a rewrite
• You can mix both APIs in the same project (capsule 07)

Next capsule: Patterns with the Functional API — reusable recipes for the most common patterns that fall out naturally with @entrypoint and @task.

Additional resources

1. Functional API Conceptual Guide — Official documentation for the Functional API
2. Graph API Conceptual Guide — Official documentation for the Graph API
3. Workflows and Agents Patterns — Official side-by-side patterns in both APIs
4. How to use the Functional API — Practical tutorial for the Functional API
5. How to build a ReAct agent from scratch — The ReAct pattern implemented with the Graph API
6. Choosing between Graph API and Functional API — Official decision guide between the APIs
7. LangGraph StateGraph Reference — Complete StateGraph reference

Module 6 — LangChain & LangGraph: From Chains to Agents