If you’re searching for a deepseek-r1 langgraph reasoning tutorial that goes beyond “hello world,” this walkthrough shows how to wire DeepSeek-R1’s chain-of-thought output into a production-grade LangGraph workflow. We’ll cover stateful reasoning loops, streaming intermediate steps to the client, and graceful degradation when the model provider hiccups.

Prerequisites

You need Python 3.10+, an OpenRouter-compatible API key (or direct DeepSeek API access), and the following packages:

pip install langgraph langchain-openai langchain-core httpx python-dotenv

Create a .env file with your credentials:

# .env
OPENROUTER_API_KEY=sk-or-v1-...
# Or if using DeepSeek directly:
# DEEPSEEK_API_KEY=sk-...

The code below assumes an OpenAI-compatible endpoint. If you’re hitting DeepSeek directly, change base_url to https://api.deepseek.com/v1 and use DEEPSEEK_API_KEY.

Minimal working graph

Start with a graph that invokes DeepSeek-R1 once and returns the full response including reasoning tokens.

# graph_basic.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

load_dotenv()

class ReasoningState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    reasoning: str | None
    final_answer: str | None

llm = ChatOpenAI(
    model="deepseek/deepseek-r1",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    temperature=0.6,
    max_tokens=8192,
    streaming=True,
)

def reason_node(state: ReasoningState) -> ReasoningState:
    response = llm.invoke(state["messages"])
    # DeepSeek-R1 returns reasoning in a separate field on some providers,
    # but OpenRouter surfaces it as part of content with " in content:
        start = content.index("")
        reasoning = content[start:end].strip()
        final = content[end + 8:].strip()
    
    return {
        "messages": [AIMessage(content=final)],
        "reasoning": reasoning,
        "final_answer": final,
    }

builder = StateGraph(ReasoningState)
builder.add_node("reason", reason_node)
builder.set_entry_point("reason")
builder.add_edge("reason", END)

graph = builder.compile()

if __name__ == "__main__":
    result = graph.invoke({
        "messages": [HumanMessage(content="Solve: 37 * 42 step by step")],
        "reasoning": None,
        "final_answer": None,
    })
    print("=== REASONING ===")
    print(result["reasoning"])
    print("\n=== ANSWER ===")
    print(result["final_answer"])

Run it:

python graph_basic.py

Expected output (truncated):

=== REASONING ===
The user wants me to multiply 37 by 42 step by step.
I'll use the standard multiplication algorithm:
37 * 42 = 37 * (40 + 2) = 37*40 + 37*2 = 1480 + 74 = 1554
Let me verify: 37 * 40 = 1480, 37 * 2 = 74, 1480 + 74 = 1554.

=== ANSWER ===
37 × 42 = 1,554

Streaming reasoning tokens

Real applications need to show reasoning as it arrives. LangGraph’s astream with stream_mode="messages" yields tokens, but DeepSeek-R1’s `“ in buffer: start = buffer.index(“”) reasoning_piece = buffer[start:end] buffer = buffer[end + 8:] yield {“type”: “reasoning”, “content”: reasoning_piece} in_think = False

Handle partial think blocks

if “” not in buffer: start = buffer.index(“ tags. Provide your final answer after the closing tag.“”“

CRITIQUE_PROMPT = “”“Review the following reasoning and answer for correctness. Identify any logical gaps, calculation errors, or unstated assumptions. Respond with a critique wrapped in , then a verdict: PASS or REVISE.”“”

def extract_think(content: str) -> tuple[str, str]: if “” in content: start = content.index(“”) return content[start:end].strip(), content[end + 8:].strip() return “”, content.strip()

def draft_node(state: ReflectState) -> ReflectState: messages = [ SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=state[“problem”]), ] response = llm.invoke(messages) reasoning, answer = extract_think(response.content) return { “draft_answer”: answer, “messages”: [AIMessage(content=f“Reasoning: {reasoning}\nAnswer: {answer}”)], “iteration”: state[“iteration”] + 1, }

def critique_node(state: ReflectState) -> ReflectState: messages = [ SystemMessage(content=CRITIQUE_PROMPT), HumanMessage(content=f“Problem: {state[‘problem’]}\n\nDraft: {state[‘draft_answer’]}”), ] response = llm.invoke(messages) reasoning, verdict = extract_think(response.content) return { “critique”: reasoning, “messages”: [AIMessage(content=f“Critique: {reasoning}\nVerdict: {verdict}”)], }

def revise_node(state: ReflectState) -> ReflectState: messages = [ SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=f“““Problem: {state[‘problem’]} Previous answer: {state[‘draft_answer’]} Critique: {state[‘critique’]} Revise your answer addressing the critique.”“”), ] response = llm.invoke(messages) reasoning, answer = extract_think(response.content) return { “revision”: answer, “draft_answer”: answer, # becomes the new draft for next iteration “messages”: [AIMessage(content=f“Revised reasoning: {reasoning}\nRevised answer: {answer}”)], }

def should_continue(state: ReflectState) -> Literal[“critique”, “revise”, END]: if state[“iteration”] >= state[“max_iterations”]: return END

Check last critique verdict

last_msg = state[“messages”][-1].content if “VERDICT: PASS” in last_msg.upper(): return END return “revise”

builder = StateGraph(ReflectState) builder.add_node(“draft”, draft_node) builder.add_node(“critique”, critique_node) builder.add_node(“revise”, revise_node)

builder.set_entry_point(“draft”) builder.add_edge(“draft”, “critique”) builder.add_conditional_edges(“critique”, should_continue, { “revise”: “revise”, END: END, }) builder.add_edge(“revise”, “critique”)

graph = builder.compile()

if name == “main”: result = graph.invoke({ “messages”: [], “problem”: “A bat and ball cost $1.10 total. The bat costs $1.00 more than the ball. How much is the ball?”, “draft_answer”: None, “critique”: None, “revision”: None, “iteration”: 0, “max_iterations”: 3, })

print(“=== FINAL ANSWER ===”) print(result.get(“revision”) or result.get(“draft_answer”)) print(f“\nIterations: {result[‘iteration’]}“)


The classic bat-and-ball problem trips up many models on the first pass. With reflection, DeepSeek-R1 typically self-corrects:

=== FINAL ANSWER === The ball costs $0.05 (5 cents). The bat costs $1.05. Total: $1.10. Difference: $1.00.

Iterations: 2


## Provider fallback and observability

In production you need fallback when your primary provider degrades. Since DeepSeek-R1 is available through multiple OpenRouter providers, you can implement a retry-with-fallback pattern at the LangGraph level.

```python
# graph_resilient.py
import os
import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

load_dotenv()

class ResilientState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    provider_used: str
    attempts: int

# Primary: DeepSeek via OpenRouter
# Fallback: DeepSeek direct (if you have key) or another reasoning model
PROVIDERS = [
    &#123;
        "name": "openrouter-deepseek-r1",
        "model": "deepseek/deepseek-r1",
        "base_url": "https://openrouter.ai/api/v1",
        "api_key": os.getenv("OPENROUTER_API_KEY"),
    &#125;,
    &#123;
        "name": "openrouter-deepseek-r1-free",
        "model": "deepseek/deepseek-r1:free",
        "base_url": "https://openrouter.ai/api/v1",
        "api_key": os.getenv("OPENROUTER_API_KEY"),
    &#125;,
]

def make_llm(provider: dict) -> ChatOpenAI:
    return ChatOpenAI(
        model=provider["model"],
        base_url=provider["base_url"],
        api_key=provider["api_key"],
        temperature=0.6,
        max_tokens=8192,
        timeout=60,
        max_retries=0,  # we handle retries ourselves
    )

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(2),
    retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)),
)
async def invoke_with_fallback(messages: list, providers: list) -> tuple[AIMessage, str]:
    last_error = None
    for provider in providers:
        llm = make_llm(provider)
        try:
            response = await llm.ainvoke(messages)
            return AIMessage(content=response.content), provider["name"]
        except Exception as e:
            last_error = e
            continue
    raise last_error or RuntimeError("All providers exhausted")

async def resilient_node(state: ResilientState) -> ResilientState:
    response, provider = await invoke_with_fallback(state["messages"], PROVIDERS)
    return &#123;
        "messages": [response],
        "provider_used": provider,
        "attempts": state.get("attempts", 0) + 1,
    &#125;

builder = StateGraph(ResilientState)
builder.add_node("resilient_reason", resilient_node)
builder.set_entry_point("resilient_reason")
builder.add_edge("resilient_reason", END)

graph = builder.compile()

async def main():
    result = await graph.ainvoke(&#123;
        "messages": [HumanMessage(content="Prove there are infinitely many primes")],
        "provider_used": "",
        "attempts": 0,
    &#125;)
    print(f"Provider: &#123;result['provider_used']&#125;")
    print(f"Attempts: &#123;result['attempts']&#125;")
    print(result["messages"][-1].content[:500] + "...")

if __name__ == "__main__":
    asyncio.run(main())

This pattern mirrors what n4n.ai handles at the gateway level — automatic fallback across 240+ models when a provider is rate-limited or degraded — but implementing it in your graph gives you application-level visibility and control over which fallbacks are acceptable for your use case.

Structured output for downstream consumers

If you need the reasoning and answer as separate structured fields (for logging, eval, or UI), add a Pydantic parser node.

# graph_structured.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from pydantic import BaseModel, Field
import operator

load_dotenv()

class ReasoningOutput(BaseModel):
    reasoning: str = Field(description="Step-by-step chain of thought")
    answer: str = Field(description="Final concise answer")
    confidence: float = Field(ge=0, le=1, description="Self-assessed confidence")

class StructuredState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    structured: ReasoningOutput | None

llm = ChatOpenAI(
    model="deepseek/deepseek-r1",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    temperature=0.6,
    max_tokens=8192,
).with_structured_output(ReasoningOutput, method="function_calling")

def structured_node(state: StructuredState) -> StructuredState:
    # Prepend instruction for structured output
    messages = [
        *state["messages"],
        HumanMessage(content="""Provide your response as a function call with:
- reasoning: your step-by-step thinking
- answer: final answer only
- confidence: 0.0 to 1.0"""),
    ]
    result = llm.invoke(messages)
    return &#123;
        "messages": [AIMessage(content=f"Reasoning: &#123;result.reasoning&#125;\nAnswer: &#123;result.answer&#125;")],
        "structured": result,
    &#125;

builder = StateGraph(StructuredState)
builder.add_node("structured", structured_node)
builder.set_entry_point("structured")
builder.add_edge("structured", END)

graph = builder.compile()

if __name__ == "__main__":
    result = graph.invoke(&#123;
        "messages": [HumanMessage(content="What is the derivative of x^2 * sin(x)?")],
        "structured": None,
    &#125;)
    out = result["structured"]
    print(f"Reasoning: &#123;out.reasoning[:200]&#125;...")
    print(f"Answer: &#123;out.answer&#125;")
    print(f"Confidence: &#123;out.confidence&#125;")

Output:

Reasoning: Use the product rule: d/dx[u*v] = u'*v + u*v'. Let u = x^2, v = sin(x). Then u' = 2x, v' = cos(x). So derivative = 2x*sin(x) + x^2*cos(x)...
Answer: 2x*sin(x) + x^2*cos(x)
Confidence: 0.95

Testing checklist

Before shipping, verify these behaviors:

Scenario	Test	Expected
Cold start	First request after deploy	< 3s to first reasoning token
Streaming	Long reasoning task	Tokens arrive steadily, no 30s gaps
Fallback	Kill primary provider	Graph completes via secondary, `provider_used` reflects fallback
Reflection loop	Known trap problem (bat/ball)	Self-corrects within `max_iterations`
Structured output	Parse `ReasoningOutput`	No validation errors, confidence calibrated

Where to go next

Evaluation: Log reasoning and answer pairs to a dataset. Grade reasoning quality separately from answer correctness.
Caching: DeepSeek-R1 reasoning is deterministic at temperature 0. Cache by prompt hash to skip repeat calls.
Budget control: Add a max_reasoning_tokens guard in the state — truncate or early-exit if the model overthinks.
Human-in-the-loop: Insert an interrupt node after critique for expert review on high-stakes tasks.

The patterns here — streaming parse, reflection loops, provider fallback, structured output — compose into a reasoning layer you can drop into any LangGraph application. Start with the minimal graph, add streaming when the UX demands it, then layer in resilience as traffic grows.