xiaohongshu
xiaohongshu
building-langgraph-agents
LangGraph development for stateful multi-agent applications, cyclic workflows, conditional routing, human-in-the-loop patterns, and persistent state management. Use for complex AI orchestration, agent coordination, and production-grade agentic systems.
Full skill instructions
LangGraph is LangChain's framework for building stateful, multi-actor applications with LLMs. It enables cyclic graphs, conditional branching, persistent state management, and human-in-the-loop workflows.
Key capabilities: StateGraph for workflow orchestration, checkpointers for persistence, prebuilt ReAct agents, streaming at multiple granularities, and LangGraph Platform for managed deployment.
This skill is loaded by backend-developer when:
langgraph in requirements.txt, pyproject.toml, or setup.py@langchain/langgraph in package.json dependenciesfrom langgraph.graph import StateGraphLANGSMITH_API_KEY, LANGGRAPH_API_URL, or LANGCHAIN_API_KEY present| Package | Description |
|---|---|
langgraph | Core StateGraph, MessageGraph, prebuilt components |
langgraph-checkpoint-postgres | PostgreSQL checkpointer for production |
langgraph-checkpoint-sqlite | SQLite checkpointer with async support |
langgraph-sdk | Python client for LangGraph Platform API |
@langchain/langgraph | TypeScript/JavaScript implementation |
Compatibility: LangGraph 0.2.x requires langchain-core>=0.3.0.
"""LangGraph Quick Start - Python"""
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# 1. Define state schema
class State(TypedDict):
messages: list[str]
count: int
# 2. Define node functions
def process(state: State) -> dict:
"""Process node - increments counter."""
return {"count": state["count"] + 1}
def respond(state: State) -> dict:
"""Response node - adds message to state."""
return {"messages": state["messages"] + [f"Processed {state['count']} times"]}
# 3. Build graph
graph = StateGraph(State)
graph.add_node("process", process)
graph.add_node("respond", respond)
graph.add_edge(START, "process")
graph.add_edge("process", "respond")
graph.add_edge("respond", END)
# 4. Compile and run
app = graph.compile()
result = app.invoke({"messages": [], "count": 0})
# {'messages': ['Processed 1 times'], 'count': 1}
"""LangGraph with ChatOpenAI and tools."""
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"Weather in {city}: 72F, sunny"
llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather]
llm_with_tools = llm.bind_tools(tools)
def agent(state: MessagesState) -> dict:
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent")
app = graph.compile()
result = app.invoke({"messages": [("user", "What's the weather in NYC?")]})
import { StateGraph, START, END, Annotation } from "@langchain/langgraph";
const State = Annotation.Root({
messages: Annotation<string[]>({
default: () => [],
reducer: (curr, update) => [...curr, ...update],
}),
count: Annotation<number>({ default: () => 0 }),
});
const process = (state: typeof State.State) => ({
count: state.count + 1,
});
const respond = (state: typeof State.State) => ({
messages: [`Processed ${state.count} times`],
});
const graph = new StateGraph(State)
.addNode("process", process)
.addNode("respond", respond)
.addEdge(START, "process")
.addEdge("process", "respond")
.addEdge("respond", END);
const app = graph.compile();
const result = await app.invoke({ messages: [], count: 0 });
The core abstraction - a directed graph where nodes read and write to shared state.
from langgraph.graph import StateGraph
graph = StateGraph(State) # State is a TypedDict
Defines the shape of data flowing through the graph.
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # With reducer
context: str # Simple overwrite
Functions that receive state and return partial updates.
def my_node(state: State) -> dict:
return {"context": f"{state['context']} - processed"}
Connections between nodes - unconditional or conditional.
# Unconditional
graph.add_edge("node_a", "node_b")
# Conditional
graph.add_conditional_edges("router", routing_function)
| Node | Purpose |
|---|---|
START | Entry point of the graph |
END | Terminal node, execution stops |
| Component | Import | Purpose |
|---|---|---|
create_react_agent | langgraph.prebuilt | Full ReAct agent with tool loop |
ToolNode | langgraph.prebuilt | Execute tools from AI messages |
tools_condition | langgraph.prebuilt | Route: tools called vs end |
MessagesState | langgraph.graph | Prebuilt chat state schema |
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, [search])
result = agent.invoke({"messages": [("user", "Search for LangGraph")]})
graph.add_edge(START, "extract")
graph.add_edge("extract", "transform")
graph.add_edge("transform", "load")
graph.add_edge("load", END)
def router(state: State) -> str:
if state["intent"] == "search":
return "search_node"
return "default_node"
graph.add_conditional_edges("classifier", router)
def check_done(state):
return END if state["done"] else "process"
graph.add_edge(START, "process")
graph.add_edge("process", "check")
graph.add_conditional_edges("check", check_done)
from langgraph.checkpoint.memory import MemorySaver
# Development - in-memory (lost on restart)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [("user", "Hi")]}, config)
For production persistence, see REFERENCE.md for PostgreSQL/SQLite setup.
# Install
pip install langgraph langgraph-checkpoint-postgres
# Verify
python -c "import langgraph; print(langgraph.__version__)"
# LangGraph Platform
langgraph init # Initialize project
langgraph dev # Start dev server
langgraph build # Build for deployment
langgraph up # Deploy with Docker
export LANGSMITH_API_KEY="ls-..."
export LANGCHAIN_TRACING_V2="true"
export OPENAI_API_KEY="sk-..."
| Error | Solution |
|---|---|
InvalidUpdateError | Node returned keys not in state schema |
GraphRecursionError | Add exit condition or increase recursion_limit |
ValueError: Checkpointer required | Add checkpointer to compile() for thread ops |
Load REFERENCE.md for:
| Agent | Use Case |
|---|---|
backend-developer | Graph implementation, state design |
deep-debugger | Execution issues, state bugs |
tech-lead-orchestrator | Multi-agent architecture |
code-reviewer | Graph logic review |
Progressive Disclosure: Start here for quick implementation. Load REFERENCE.md for comprehensive patterns, production setup, and advanced features.
xiaohongshu
technical spec
product ux expert
database patterns
Conduct multi-agent task orchestration and workflow coordination.
Initialize project with Conductor artifacts (product definition,
Expert in web animations, transitions, and motion design using Framer Motion and CSS
Creates Mermaid and ASCII diagrams for flowcharts, architecture, ERDs, state machines, mindmaps, and more. Use when user mentions diagram, flowchart, mermaid, ASCII diagram, text diagram, terminal diagram, visualize, C4, mindmap, architecture diagram, sequence diagram, ERD, or needs visual docume...
PostgreSQL bindings for H3 hexagonal grid system. Use when working with H3 cells in Postgres, including spatial indexing, geometry/geography integration, and raster analysis.
Context-Driven Development skill for projects using Conductor. Use this skill when you detect a `conductor/` directory in the project, when working on tasks defined in a `plan.md` file, or when the user asks about tracks, specs, or plans. Automatically applies TDD workflow, tracks task completion...
Display project status, active tracks, and next actions
Official Stakpak application containerization standard operating procedure, a step-by-step guidline to properly dockerize applications. This is a rule book curated by the Stakpak Team.
Generate, edit, and beat-sync AI video with leading models in one workspace.
The world's fastest calendar for remote work
Transform Your Design with AI Designer by ImgCreator.ai
Revolutionizing Video Production with AI-Powered Creativity
Extend an image past the frame and let AI fill the new aspect ratio.
Discover your celebrity doppelgänger with StarByFace!
ChainClarity explains 700+ crypto whitepapers in plain English, with layered summaries, comparisons, research tools, alerts, and a $4.99 Pro plan.
Opus.ai: Revolutionize Your Web Experience