Graph API vs Functional API choice
Use the Graph API if you prefer to define your agent as a graph of nodes and edges. Use the Functional API if you prefer to define your agent as a single function.
LangChain · LangGraph · all subjects
graph api and fundamentals
33 notes, read out of this brain and free to use. Each one was extracted from a source and is re-checked against its exam.
StateGraph basic structure
To build an agent with StateGraph: create instance with state class, add nodes with add_node(), add edges with add_edge() and add_conditional_edges(), then compile() to produce the runnable agent.
Functional API uses @task and @entrypoint decorators
In the Functional API, mark functions with @task decorator to mark them as tasks that can be executed as part of the agent. Mark the main function with @entrypoint() decorator. Tasks can be called synchronously or asynchronously within the entrypoint function.
Functional API control flow instead of nodes and edges
In the Functional API, instead of defining nodes and edges explicitly, write standard control flow logic (loops, conditionals) within a single entrypoint function.
Graph API calculator agent example in Python
Complete working example: Define MessagesState with Annotated messages field using operator.add. Create llm_call node that invokes model with tools. Create tool_node that processes tool calls. Use should_continue conditional edge to route based on tool_calls presence. Connect START → llm_call → conditional edge to [tool_node, END] and tool_node → llm_call. Compile and invoke with StateGraph.
Functional API calculator agent example in Python
Complete working example: Define call_llm task decorated with @task that invokes modelWithTools. Define call_tool task that takes a ToolCall and invokes the tool. Define agent function decorated with @entrypoint that calls call_llm, loops while tool_calls exist, executes tools via call_tool tasks, accumulates messages with add_messages, and returns final messages.
Task.result() returns value synchronously in Functional API
In the Functional API, call .result() on a task future to get the synchronous result, e.g., model_response = call_llm(messages).result().
Workflows vs agents definition
Workflows have predetermined code paths and are designed to operate in a certain order. Agents are dynamic and define their own processes and tool usage.
Prompt chaining pattern
Prompt chaining is when each LLM call processes the output of the previous call. It is often used for performing well-defined tasks that can be broken down into smaller, verifiable steps. Examples include translating documents into different languages and verifying generated content for consistency.
Parallelization pattern uses
Parallelization is commonly used to split up subtasks and run them in parallel to increase speed, or run tasks multiple times to check for different outputs to increase confidence. Examples include running one subtask that processes a document for keywords and a second subtask to check for formatting errors, or running a task multiple times that scores a document for accuracy based on different criteria like the number of citations, the number of sources used, and the quality of the sources.
Routing pattern definition
Routing workflows process inputs and then direct them to context-specific tasks. This allows you to define specialized flows for complex tasks. For example, a workflow built to answer product related questions might process the type of question first, and then route the request to specific processes for pricing, refunds, returns, etc.
Orchestrator-worker pattern
In an orchestrator-worker configuration, the orchestrator breaks down tasks into subtasks, delegates subtasks to workers, and synthesizes worker outputs into a final result. Orchestrator-worker workflows provide more flexibility and are often used when subtasks cannot be predefined the way they can with parallelization. This is common with workflows that write code or need to update content across multiple files.
Send API for dynamic worker creation
The Send API lets you dynamically create worker nodes and send them specific inputs. Each worker has its own state, and all worker outputs are written to a shared state key that is accessible to the orchestrator graph. This gives the orchestrator access to all worker output and allows it to synthesize them into a final output.
Evaluator-optimizer pattern
In evaluator-optimizer workflows, one LLM call creates a response and the other evaluates that response. If the evaluator or a human-in-the-loop determines the response needs refinement, feedback is provided and the response is recreated. This loop continues until an acceptable response is generated. Evaluator-optimizer workflows are commonly used when there is particular success criteria for a task, but iteration is required to meet that criteria.
Agents operate in continuous feedback loops
Agents are typically implemented as an LLM performing actions using tools. They operate in continuous feedback loops and are used in situations where problems and solutions are unpredictable. Agents have more autonomy than workflows and can make decisions about the tools they use and how to solve problems. You can still define the available toolset and guidelines for how agents behave.
Prompt chaining Graph API example
from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from IPython.display import Image, display class State(TypedDict): topic: str joke: str improved_joke: str final_joke: str def generate_joke(state: State): msg = llm.invoke(f"Write a short joke about {state['topic']}") return {"joke": msg.content} def check_punchline(state: State): if "?" in state["joke"] or "!" in state["joke"]: return "Pass" return "Fail" def improve_joke(state: State): msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}") return {"improved_joke": msg.content} def polish_joke(state: State): msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}") return {"final_joke": msg.content} workflow = StateGraph(State) workflow.add_node("generate_joke", generate_joke) workflow.add_node("improve_joke", improve_joke) workflow.add_node("polish_joke", polish_joke) workflow.add_edge(START, "generate_joke") workflow.add_conditional_edges( "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END} ) workflow.add_edge("improve_joke", "polish_joke") workflow.add_edge("polish_joke", END) chain = workflow.compile()
Parallelization Graph API example
from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): topic: str joke: str story: str poem: str combined_output: str def call_llm_1(state: State): msg = llm.invoke(f"Write a joke about {state['topic']}") return {"joke": msg.content} def call_llm_2(state: State): msg = llm.invoke(f"Write a story about {state['topic']}") return {"story": msg.content} def call_llm_3(state: State): msg = llm.invoke(f"Write a poem about {state['topic']}") return {"poem": msg.content} def aggregator(state: State): combined = f"Here's a story, joke, and poem about {state['topic']}!\n\n" combined += f"STORY:\n{state['story']}\n\n" combined += f"JOKE:\n{state['joke']}\n\n" combined += f"POEM:\n{state['poem']}" return {"combined_output": combined} parallel_builder = StateGraph(State) parallel_builder.add_node("call_llm_1", call_llm_1) parallel_builder.add_node("call_llm_2", call_llm_2) parallel_builder.add_node("call_llm_3", call_llm_3) parallel_builder.add_node("aggregator", aggregator) parallel_builder.add_edge(START, "call_llm_1") parallel_builder.add_edge(START, "call_llm_2") parallel_builder.add_edge(START, "call_llm_3") parallel_builder.add_edge("call_llm_1", "aggregator") parallel_builder.add_edge("call_llm_2", "aggregator") parallel_builder.add_edge("call_llm_3", "aggregator") parallel_builder.add_edge("aggregator", END) parallel_workflow = parallel_builder.compile()
Routing Graph API example
from typing_extensions import Literal from pydantic import BaseModel, Field from langchain.messages import HumanMessage, SystemMessage from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict class Route(BaseModel): step: Literal["poem", "story", "joke"] = Field(None, description="The next step in the routing process") router = llm.with_structured_output(Route) class State(TypedDict): input: str decision: str output: str def llm_call_1(state: State): result = llm.invoke(state["input"]) return {"output": result.content} def llm_call_2(state: State): result = llm.invoke(state["input"]) return {"output": result.content} def llm_call_3(state: State): result = llm.invoke(state["input"]) return {"output": result.content} def llm_call_router(state: State): decision = router.invoke([ SystemMessage(content="Route the input to story, joke, or poem based on the user's request."), HumanMessage(content=state["input"]), ]) return {"decision": decision.step} def route_decision(state: State): if state["decision"] == "story": return "llm_call_1" elif state["decision"] == "joke": return "llm_call_2" elif state["decision"] == "poem": return "llm_call_3" router_builder = StateGraph(State) router_builder.add_node("llm_call_1", llm_call_1) router_builder.add_node("llm_call_2", llm_call_2) router_builder.add_node("llm_call_3", llm_call_3) router_builder.add_node("llm_call_router", llm_call_router) router_builder.add_edge(START, "llm_call_router") router_builder.add_conditional_edges( "llm_call_router", route_decision, {"llm_call_1": "llm_call_1", "llm_call_2": "llm_call_2", "llm_call_3": "llm_call_3"}, ) router_builder.add_edge("llm_call_1", END) router_builder.add_edge("llm_call_2", END) router_builder.add_edge("llm_call_3", END) router_workflow = router_builder.compile()
Orchestrator-worker with Send API example
from typing import Annotated, List import operator from langgraph.types import Send from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class Section(BaseModel): name: str = Field(description="Name for this section of the report.") description: str = Field(description="Brief overview of the main topics and concepts to be covered in this section.") class State(TypedDict): topic: str sections: list[Section] completed_sections: Annotated[list, operator.add] final_report: str class WorkerState(TypedDict): section: Section completed_sections: Annotated[list, operator.add] def orchestrator(state: State): report_sections = planner.invoke([ SystemMessage(content="Generate a plan for the report."), HumanMessage(content=f"Here is the report topic: {state['topic']}"), ]) return {"sections": report_sections.sections} def llm_call(state: WorkerState): section = llm.invoke([ SystemMessage(content="Write a report section following the provided name and description. Include no preamble for each section. Use markdown formatting."), HumanMessage(content=f"Here is the section name: {state['section'].name} and description: {state['section'].description}"), ]) return {"completed_sections": [section.content]} def synthesizer(state: State): completed_sections = state["completed_sections"] completed_report_sections = "\n\n---\n\n".join(completed_sections) return {"final_report": completed_report_sections} def assign_workers(state: State): return [Send("llm_call", {"section": s}) for s in state["sections"]] orchestrator_worker_builder = StateGraph(State) orchestrator_worker_builder.add_node("orchestrator", orchestrator) orchestrator_worker_builder.add_node("llm_call", llm_call) orchestrator_worker_builder.add_node("synthesizer", synthesizer) orchestrator_worker_builder.add_edge(START, "orchestrator") orchestrator_worker_builder.add_conditional_edges("orchestrator", assign_workers, ["llm_call"]) orchestrator_worker_builder.add_edge("llm_call", "synthesizer") orchestrator_worker_builder.add_edge("synthesizer", END) orchestrator_worker = orchestrator_worker_builder.compile()
Evaluator-optimizer Graph API example
from typing_extensions import TypedDict, Literal from pydantic import BaseModel, Field from langgraph.graph import StateGraph, START, END class State(TypedDict): joke: str topic: str feedback: str funny_or_not: str class Feedback(BaseModel): grade: Literal["funny", "not funny"] = Field(description="Decide if the joke is funny or not.") feedback: str = Field(description="If the joke is not funny, provide feedback on how to improve it.") evaluator = llm.with_structured_output(Feedback) def llm_call_generator(state: State): if state.get("feedback"): msg = llm.invoke(f"Write a joke about {state['topic']} but take into account the feedback: {state['feedback']}") else: msg = llm.invoke(f"Write a joke about {state['topic']}") return {"joke": msg.content} def llm_call_evaluator(state: State): grade = evaluator.invoke(f"Grade the joke {state['joke']}") return {"funny_or_not": grade.grade, "feedback": grade.feedback} def route_joke(state: State): if state["funny_or_not"] == "funny": return "Accepted" elif state["funny_or_not"] == "not funny": return "Rejected + Feedback" optimizer_builder = StateGraph(State) optimizer_builder.add_node("llm_call_generator", llm_call_generator) optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator) optimizer_builder.add_edge(START, "llm_call_generator") optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator") optimizer_builder.add_conditional_edges( "llm_call_evaluator", route_joke, {"Accepted": END, "Rejected + Feedback": "llm_call_generator"}, ) optimizer_workflow = optimizer_builder.compile()
Agent with tools Graph API example
from langgraph.graph import MessagesState, StateGraph, START, END from langchain.messages import SystemMessage, HumanMessage, ToolMessage from typing_extensions import Literal def llm_call(state: MessagesState): return { "messages": [ llm_with_tools.invoke( [SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")] + state["messages"] ) ] } def tool_node(state: MessagesState): result = [] for tool_call in state["messages"][-1].tool_calls: tool = tools_by_name[tool_call["name"]] observation = tool.invoke(tool_call["args"]) result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"])) return {"messages": result} def should_continue(state: MessagesState) -> Literal["tool_node", END]: messages = state["messages"] last_message = messages[-1] if last_message.tool_calls: return "tool_node" return END agent_builder = StateGraph(MessagesState) agent_builder.add_node("llm_call", llm_call) agent_builder.add_node("tool_node", tool_node) agent_builder.add_edge(START, "llm_call") agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END]) agent_builder.add_edge("tool_node", "llm_call") agent = agent_builder.compile()
StateGraph API for graph creation
StateGraph is used to define a graph that operates on a specific state. After creating a StateGraph instance, use add_node() (Python) or addNode() (JavaScript) to populate the graph with nodes, then call compile() to create the executable graph.
Input and output schemas in StateGraph
StateGraph allows defining distinct input and output schemas separate from the internal state schema. The input schema ensures the provided input matches the expected structure, while the output schema filters the internal data to return only relevant information. An internal schema is still used for communication between nodes. Specify them with input_schema and output_schema parameters (Python) or input and output fields in StateGraph configuration (JavaScript).
Runtime configuration in graphs
Runtime configuration allows configuring a graph at invocation time without polluting the graph state. To add runtime configuration: (1) Specify a schema for configuration; (2) Add the configuration to the function signature for nodes or conditional edges via runtime parameter; (3) Pass the configuration into the graph via context parameter. In Python use Runtime[ContextSchema] parameter, in JavaScript use config parameter with configurable object.
set_node_defaults for graph-wide node configuration
Use set_node_defaults() to set retry_policy, timeout, cache_policy, or error_handler once for every node in a graph instead of repeating them on each add_node() call. Per-node values always win over defaults. Defaults are applied at StateGraph.compile() time. retry_policy and timeout defaults apply to every node including error-handler nodes. cache_policy and error_handler defaults apply only to regular nodes. Requires langgraph>=1.2.
Access execution info in a node
Access execution identity and retry information via runtime.execution_info to get thread_id (str|None), run_id (str|None), checkpoint_id (str), checkpoint_ns (str), task_id (str), node_attempt (int, 1-indexed), and node_first_attempt_time (float|None as Unix timestamp). Requires deepagents>=0.5.0 or langgraph>=1.1.5.
runtime.executionInfo attributes
The runtime.executionInfo object provides execution identity and retry information with the following attributes: threadId (string | undefined) - Thread ID for the current execution; runId (string | undefined) - Run ID for the current execution; checkpointId (string) - Checkpoint ID for the current execution; checkpointNs (string) - Checkpoint namespace for the current execution; taskId (string) - Task ID for the current execution; nodeAttempt (number) - Current execution attempt number (1-indexed); nodeFirstAttemptTime (number | undefined) - Unix timestamp in seconds of when the first attempt started, stays the same across retries.
Access thread and run IDs in nodes
Inside a node function, use runtime.executionInfo to access thread ID, run ID, and other identity fields. Example: const info = runtime.executionInfo; console.log(`Thread: ${info.threadId}, Run: ${info.runId}`);
Why sequences are important in LangGraph
Splitting application steps into sequences with LangGraph enables: checkpointing state between node executions, resuming interruptions in human-in-the-loop workflows, rewinding and branching executions with time travel features, streaming execution steps, and visualization/debugging in Studio.
Control recursion limit
Set recursion_limit in the config when invoking the graph to limit the number of supersteps executed. This raises GraphRecursionError after the limit is reached. Example Python: graph.invoke(inputs, {'recursion_limit': 3}). Example TypeScript: await graph.invoke(inputs, {recursionLimit: 3});
RemainingSteps annotation for step tracking
Use the RemainingSteps annotation to track remaining steps until recursion limit. This creates a ManagedValue channel that exists for the duration of a graph run. Allows graceful termination before hitting recursion limit. Example: remaining_steps: RemainingSteps in state.
Graph compilation and validation
Compile the graph after defining nodes and edges using .compile(). This performs basic checks on graph structure (e.g., identifying orphaned nodes). If using persistence, pass a checkpointer to compile().
Visualization of graphs
Visualize graphs using graph.get_graph().draw_mermaid_png() (Python) or graph.getGraphAsync().drawMermaidPng() (TypeScript). For Python Jupyter notebooks, use display(Image(graph.get_graph().draw_mermaid_png())). This creates a visual representation of nodes and edges.