LangGraph Basics

LangGraph is a framework for building stateful AI agents as directed graphs. For an Angular developer building AI-powered applications, the sections below cover how LangGraph agents work and why injectAgent() is the natural bridge between your frontend and your agent backend.

Note: Why graphs?

Graphs give you explicit control over agent behavior. Instead of a black-box prompt-and-pray approach, you define exactly how your agent reasons, when it calls tools, and where it pauses for human input. Every step is visible, testable, and debuggable.

The Core Concepts

A LangGraph agent has three building blocks:

Nodes — Functions That Do Work

A node is a Python function that receives the current state, does something, and returns updated state. Every node has the same signature:

def my_node(state: State, config: RunnableConfig) -> dict:
    # Read from state
    messages = state["messages"]
 
    # Do work (call LLM, query DB, invoke tool)
    response = llm.invoke(messages)
 
    # Return state updates (merged into existing state)
    return {"messages": [response]}

Tip: Node return values

Nodes do not replace state — they return updates that get merged into the existing state. For lists like messages, LangGraph uses reducers (like operator.add) to accumulate entries instead of overwriting.

Edges — Connections Between Nodes

Edges define the execution flow. There are two types:

Normal edges — always route to the next node:

builder.add_edge(START, "call_model")   # Start → call_model
builder.add_edge("call_model", END)     # call_model → End

Conditional edges — route based on state:

def should_continue(state: State) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"      # Agent wants to use a tool
    return END              # Agent is done, return response
 
builder.add_conditional_edges("call_model", should_continue)

State — The Shared Memory

All nodes read from and write to a shared state object. You define its shape as a Python TypedDict:

from typing_extensions import TypedDict, Annotated
from operator import add
 
class State(TypedDict):
    messages: Annotated[list, add]   # Accumulates messages
    plan: list[str]                   # Agent's current plan
    results: dict                     # Tool results

This state is exactly what injectAgent() exposes to your Angular app through Signals.

Building Your First Agent

The simplest possible agent — a chat model that takes messages and responds:

from langgraph.graph import END, START, MessagesState, StateGraph
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-5-mini")
 
def call_model(state: MessagesState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
 
# Build the graph: START → call_model → END
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
 
graph = builder.compile()

Agent Patterns

LangGraph shines in the patterns you can build. Each one maps to specific injectAgent() signals.

Pattern 1: ReAct Agent (Tool Calling)

The agent reasons, decides to call a tool, observes the result, and loops until it has an answer.

from langgraph.prebuilt import ToolNode
 
@tool
def search_docs(query: str) -> str:
    """Search the knowledge base."""
    return vector_store.similarity_search(query)
 
tools = [search_docs]
 
def call_model(state: State) -> dict:
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}
 
def should_continue(state: State) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END
 
builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue)
builder.add_edge("tools", "model")  # Loop back after tool execution
 
graph = builder.compile()

Angular connection: Track tool execution in real-time:

const agent = injectAgent();
 
// Watch tools execute
const activeTools = computed(() =>
  agent.toolCalls().filter((tool) => tool.status === 'running')
);
const completedTools = computed(() =>
  agent.toolCalls().filter((tool) => tool.status === 'complete')
);

Pattern 2: Human-in-the-Loop (Approval)

The agent proposes an action and pauses. Your Angular UI shows an approval dialog. The user decides, and the agent resumes.

from langgraph.types import interrupt
 
def propose_action(state: State) -> dict:
    action = llm.invoke(state["messages"])
    # Pause execution — Angular will show approval UI
    approval = interrupt({
        "action": "send_email",
        "to": "client@example.com",
        "body": action.content,
    })
    return {
        "pending_action": action.content,
        "approval": approval,
    }
 
def execute_action(state: State) -> dict:
    # Only runs after human approves
    if not state.get("approval", {}).get("approved"):
        return {"messages": [{"role": "assistant", "content": "Email cancelled."}]}
    send_email(state["pending_action"])
    return {"messages": [{"role": "assistant", "content": "Email sent."}]}

Angular connection: The interrupt surfaces automatically:

const agent = injectAgent();
 
// Show approval UI when agent pauses
const pendingAction = computed(() => agent.interrupt());
 
// User clicks approve → resume the agent
approve() {
  agent.submit({ resume: { approved: true } });
}

Pattern 3: Multi-Agent Orchestration

A supervisor agent delegates work to specialist sub-agents. Each sub-agent is its own graph.

def supervisor(state: State) -> dict:
    routing = llm.invoke([
        {"role": "system", "content": "Route to: researcher, analyst, or writer"},
        *state["messages"]
    ])
    return {"next_agent": routing.tool_calls[0].args["agent"]}
 
builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher_subgraph)
builder.add_node("analyst", analyst_subgraph)
builder.add_conditional_edges("supervisor", lambda s: s["next_agent"])

Angular connection: Track delegated work through dedicated subagent signals when delegation is exposed as tool calls. Configure once in app.config.ts:

provideAgent({
  apiUrl: '...',
  assistantId: 'orchestrator',
  subagentToolNames: ['task'],
});

Then consume from any component:

const orchestrator = injectAgent();
 
const workers = computed(() => [...orchestrator.subagents().values()]);
const workerCount = computed(() => workers().length);

Pattern 4: Persistent Conversations

A conversation survives page refreshes, browser restarts, and server deployments when the backend retains its checkpoints in durable storage and the client restores the same thread ID. The ID identifies the conversation; it does not make an in-memory checkpointer durable. When embedding a graph in your own process, keep every invocation inside the saver's connection scope. The example assumes your application supplies DATABASE_URL, builder, and user_message. LangGraph API deployments use the server's checkpointer; see Persistence.

from langgraph.checkpoint.postgres import PostgresSaver
 
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    checkpointer.setup()  # Initialize the checkpoint schema before first use.
    graph = builder.compile(checkpointer=checkpointer)
 
    # Each thread_id is a persistent conversation.
    result = graph.invoke(
        {"messages": [user_message]},
        config={"configurable": {"thread_id": "user_123_session"}},
    )

Angular connection: Thread persistence is configured in the provider:

// app.config.ts
provideAgent({
  apiUrl: '...',
  assistantId: 'chat',
  threadId: signal(localStorage.getItem('threadId')),
  onThreadId: (id) => localStorage.setItem('threadId', id),
});
 
// component.ts
const chat = injectAgent();
 
// User returns tomorrow — same thread, full history restored
// No code needed — the adapter handles it

How `injectAgent()` Bridges the Gap

This is why injectAgent() is the natural Angular companion for LangGraph.

Your Angular App

Calls submit({ message: text }) to send user input

`injectAgent()`

Passes input to the transport layer

FetchStreamTransport

Sends HTTP POST to LangGraph Platform, opens SSE connection

LangGraph Platform

Executes graph nodes, calls tools, streams SSE events back

FetchStreamTransport

Parses SSE chunks into typed stream events

`injectAgent()`

Converts BehaviorSubjects to Angular Signals via toSignal()

Your Angular App

Templates re-render automatically via OnPush change detection

Tip: Zero configuration streaming

You do not configure SSE, parse events, manage WebSocket connections, or handle reconnection. injectAgent() does all of that. You call submit() and read Signals — that is the entire API surface for your Angular code.

Graph API vs Functional API

LangGraph offers two ways to define agents:

Graph API (recommended for most cases):

builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
graph = builder.compile()

Functional API (for simpler workflows):

from langgraph.func import entrypoint
 
@entrypoint()
async def agent(messages):
    response = await call_model(messages)
    return response

Both APIs produce the same output and work identically with injectAgent(). Choose the Graph API when you need conditional routing, subgraphs, or interrupts. Choose the Functional API for simple, linear workflows.