LangGraph Agent Design Patterns Part 1: Prompt‑Chain, Router, and Parallel Modes

Prompt‑Chain Mode

Decomposes a complex task into a series of simpler, ordered steps. Each step invokes a large model and passes its output to the next step. Gate nodes act as quality checkpoints that decide whether to continue, branch, or terminate the workflow.

Decompose complexity – splitting a large task into manageable sub‑tasks reduces the cognitive load of each model call.

Improve accuracy and controllability – intermediate checks produce more reliable outputs than a single model invocation.

Concrete example: a multi‑stage joke‑generation workflow.

State definition records the topic and intermediate results (joke, improved_joke, final_joke).

Node functions generate_joke – first model call creates a short joke based on the topic. check_punchline – gate that verifies the joke contains a question mark or exclamation mark; returns "Fail" to trigger improvement or "Pass" to finish. improve_joke – second call adds wordplay to make the joke funnier. polish_joke – third call adds a surprising twist and final polishing.

Graph construction

from typing import TypedDict
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import END, START, StateGraph
from dotenv import load_dotenv

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str

# node definitions omitted for brevity (see description above)

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()

state = chain.invoke({"topic": "小猫"})
print(state["joke"])
if "improved_joke" in state:
    print(state["improved_joke"])
    print(state["final_joke"])

Router Mode

Classifies incoming requests and directs them to specialized downstream nodes. A router node makes the classification decision; conditional edges route the flow to the appropriate processing branch. The overall success depends heavily on the accuracy of the classification step.

Customer‑service routing – separate general FAQs, refund requests, and technical support into distinct workflows.

Resource‑cost optimization – route simple queries to a lightweight model (e.g., Qwen3‑8B) and complex requests to a more powerful model (e.g., DeepSeek‑671B).

Concrete example: a content generator that outputs a story, a joke, or a poem based on user input.

State definition records the raw input, the router decision, and the final output.

Classification schema uses a structured‑output TypedDict with a literal field response_format that can be "story", "joke" or "poetry".

Router node invokes the model with a system prompt that asks for one of the three categories and returns the chosen label.

Downstream nodes generate_story, generate_joke, and generate_poetry each call the model to produce the requested content.

Routing logic maps the decision label to the corresponding node via a conditional‑edge function.

from typing import TypedDict, Literal
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from langchain.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

class State(TypedDict):
    input: str
    decision: str
    output: str

class Classification(TypedDict):
    response_format: Literal["story", "joke", "poetry"]

def llm_call_router(state: State):
    structured_llm = llm.with_structured_output(Classification)
    response = structured_llm.invoke([
        SystemMessage(content="你是一个分类路由，根据用户的输入进行分类，分类结果是story, joke, poetry三者中的一种"),
        HumanMessage(content=state["input"])
    ])
    return {"decision": response["response_format"]}

def generate_story(state: State):
    result = llm.invoke(state["input"])
    return {"output": result.content}

def generate_joke(state: State):
    result = llm.invoke(state["input"])
    return {"output": result.content}

def generate_poetry(state: State):
    result = llm.invoke(state["input"])
    return {"output": result.content}

def route_decision(state: State):
    if state["decision"] == "story":
        return "llm_story"
    if state["decision"] == "joke":
        return "llm_joke"
    if state["decision"] == "poetry":
        return "llm_poetry"

router = StateGraph(State)
router.add_node("llm_story", generate_story)
router.add_node("llm_joke", generate_joke)
router.add_node("llm_poetry", generate_poetry)
router.add_node("llm_call_router", llm_call_router)
router.add_edge(START, "llm_call_router")
router.add_conditional_edges("llm_call_router", route_decision, {
    "llm_story": "llm_story",
    "llm_joke": "llm_joke",
    "llm_poetry": "llm_poetry"
})
router.add_edge("llm_story", END)
router.add_edge("llm_joke", END)
router.add_edge("llm_poetry", END)
workflow = router.compile()

result = workflow.invoke({"input": "给我写一个关于苍井空的笑话"})
print(result["output"])

Parallel Mode

Runs multiple sub‑tasks simultaneously and aggregates their results, dramatically improving throughput compared with sequential Prompt‑Chain execution.

Segmentation – split a complex task into independent sub‑tasks (e.g., evaluate factual accuracy, logical coherence, and style) and combine the scores.

Voting – run the same task multiple times with varied prompts or configurations, then select the majority or consensus output (e.g., multi‑model security‑risk assessment).

Concrete example: generate a joke, a story, and a poem in parallel for a given topic and then aggregate them into a single collection.

State definition holds the topic and the three generated pieces plus the combined output.

Node functions generate_joke, generate_story, and generate_poetry each invoke the model with a prompt that references the shared topic.

Aggregator node concatenates the three pieces into a formatted collection.

Graph construction connects the START node to all three generation nodes, then connects each generation node to the aggregator, and finally connects the aggregator to END, enabling true parallel execution.

from typing import TypedDict
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, START, END
from dotenv import load_dotenv

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

class State(TypedDict):
    topic: str
    joke: str
    story: str
    poetry: str
    combined_output: str

def generate_joke(state: State):
    msg = llm.invoke(f"写一个关于{state['topic']}的笑话")
    return {"joke": msg.content}

def generate_story(state: State):
    msg = llm.invoke(f"写一个关于{state['topic']}的故事")
    return {"story": msg.content}

def generate_poetry(state: State):
    msg = llm.invoke(f"写一个关于{state['topic']}的诗歌")
    return {"poetry": msg.content}

def aggregator(state: State):
    combined = f"这是一个关于 {state['topic']} 的故事、笑话和诗歌的合集

"
    combined += f"故事
{state['story']}

"
    combined += f"笑话
{state['joke']}

"
    combined += f"诗歌
{state['poetry']}

"
    return {"combined_output": combined}

parallel = StateGraph(State)
parallel.add_node("generate_joke", generate_joke)
parallel.add_node("generate_story", generate_story)
parallel.add_node("generate_poetry", generate_poetry)
parallel.add_node("aggregator", aggregator)
parallel.add_edge(START, "generate_joke")
parallel.add_edge(START, "generate_story")
parallel.add_edge(START, "generate_poetry")
parallel.add_edge("generate_joke", "aggregator")
parallel.add_edge("generate_story", "aggregator")
parallel.add_edge("generate_poetry", "aggregator")
parallel.add_edge("aggregator", END)
workflow = parallel.compile()

state = workflow.invoke({"topic": "pgone 与 李小璐"})
print(state["combined_output"])