298 lines
7.2 KiB
Markdown
298 lines
7.2 KiB
Markdown
---
|
|
name: LangGraph Patterns Expert
|
|
description: Build production-grade agentic workflows with LangGraph using graph-based orchestration, state machines, human-in-the-loop, and advanced control flow
|
|
version: 1.0.0
|
|
---
|
|
|
|
# LangGraph Patterns Expert Skill
|
|
|
|
## Purpose
|
|
Master LangGraph for building production-ready AI agents with fine-grained control, checkpointing, streaming, and complex state management.
|
|
|
|
## Core Philosophy
|
|
|
|
**LangGraph is:** An orchestration framework with both declarative and imperative APIs focused on control and durability for production agents.
|
|
|
|
**Not:** High-level abstractions that hide complexity - instead provides building blocks for full control.
|
|
|
|
**Migration:** LangGraph replaces legacy AgentExecutor - migrate all old code.
|
|
|
|
## The Six Production Features
|
|
|
|
1. **Parallelization** - Run multiple nodes concurrently
|
|
2. **Streaming** - Real-time partial outputs
|
|
3. **Checkpointing** - Pause/resume execution
|
|
4. **Human-in-the-Loop** - Approval/correction workflows
|
|
5. **Tracing** - Observability and debugging
|
|
6. **Task Queue** - Asynchronous job processing
|
|
|
|
## Graph-Based Architecture
|
|
|
|
```python
|
|
from langgraph.graph import StateGraph, END
|
|
|
|
# Define state
|
|
class AgentState(TypedDict):
|
|
messages: Annotated[list, add_messages]
|
|
next_action: str
|
|
|
|
# Create graph
|
|
graph = StateGraph(AgentState)
|
|
|
|
# Add nodes
|
|
graph.add_node("analyze", analyze_node)
|
|
graph.add_node("execute", execute_node)
|
|
graph.add_node("verify", verify_node)
|
|
|
|
# Define edges
|
|
graph.add_edge("analyze", "execute")
|
|
graph.add_conditional_edges(
|
|
"execute",
|
|
should_verify,
|
|
{"yes": "verify", "no": END}
|
|
)
|
|
|
|
# Compile
|
|
app = graph.compile()
|
|
```
|
|
|
|
## Core Patterns
|
|
|
|
### Pattern 1: Agent with Tools
|
|
```python
|
|
from langgraph.prebuilt import create_react_agent
|
|
|
|
tools = [search_tool, calculator_tool, db_query_tool]
|
|
|
|
agent = create_react_agent(
|
|
model=llm,
|
|
tools=tools,
|
|
checkpointer=MemorySaver()
|
|
)
|
|
|
|
# Run with streaming
|
|
for chunk in agent.stream({"messages": [("user", "Analyze sales data")]}):
|
|
print(chunk)
|
|
```
|
|
|
|
### Pattern 2: Multi-Agent Collaboration
|
|
```python
|
|
# Supervisor coordinates specialist agents
|
|
supervisor_graph = StateGraph(SupervisorState)
|
|
|
|
supervisor_graph.add_node("supervisor", supervisor_node)
|
|
supervisor_graph.add_node("researcher", researcher_agent)
|
|
supervisor_graph.add_node("analyst", analyst_agent)
|
|
supervisor_graph.add_node("writer", writer_agent)
|
|
|
|
# Supervisor routes to specialists
|
|
supervisor_graph.add_conditional_edges(
|
|
"supervisor",
|
|
route_to_agent,
|
|
{
|
|
"research": "researcher",
|
|
"analyze": "analyst",
|
|
"write": "writer",
|
|
"finish": END
|
|
}
|
|
)
|
|
```
|
|
|
|
### Pattern 3: Human-in-the-Loop
|
|
```python
|
|
from langgraph.checkpoint.sqlite import SqliteSaver
|
|
|
|
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
|
|
|
|
graph = StateGraph(State)
|
|
graph.add_node("propose_action", propose)
|
|
graph.add_node("human_approval", interrupt()) # Pauses here
|
|
graph.add_node("execute_action", execute)
|
|
|
|
app = graph.compile(checkpointer=checkpointer)
|
|
|
|
# Run until human input needed
|
|
result = app.invoke(input, config={"configurable": {"thread_id": "123"}})
|
|
|
|
# Human reviews, then resume
|
|
app.invoke(None, config={"configurable": {"thread_id": "123"}})
|
|
```
|
|
|
|
## State Management
|
|
|
|
### Short-Term Memory (Session)
|
|
```python
|
|
class ConversationState(TypedDict):
|
|
messages: Annotated[list, add_messages]
|
|
context: dict
|
|
|
|
checkpointer = MemorySaver()
|
|
app = graph.compile(checkpointer=checkpointer)
|
|
|
|
# Maintains context across turns
|
|
config = {"configurable": {"thread_id": "user_123"}}
|
|
app.invoke({"messages": [("user", "Hello")]}, config)
|
|
app.invoke({"messages": [("user", "What did I just say?")]}, config)
|
|
```
|
|
|
|
### Long-Term Memory (Persistent)
|
|
```python
|
|
from langgraph.checkpoint.postgres import PostgresSaver
|
|
|
|
checkpointer = PostgresSaver.from_conn_string(db_url)
|
|
|
|
# Persists across sessions
|
|
app = graph.compile(checkpointer=checkpointer)
|
|
```
|
|
|
|
## Advanced Control Flow
|
|
|
|
### Conditional Routing
|
|
```python
|
|
def route_next(state):
|
|
if state["confidence"] > 0.9:
|
|
return "approve"
|
|
elif state["confidence"] > 0.5:
|
|
return "review"
|
|
else:
|
|
return "reject"
|
|
|
|
graph.add_conditional_edges(
|
|
"classifier",
|
|
route_next,
|
|
{
|
|
"approve": "auto_approve",
|
|
"review": "human_review",
|
|
"reject": "reject_node"
|
|
}
|
|
)
|
|
```
|
|
|
|
### Cycles and Loops
|
|
```python
|
|
def should_continue(state):
|
|
if state["iterations"] < 3 and not state["success"]:
|
|
return "retry"
|
|
return "finish"
|
|
|
|
graph.add_conditional_edges(
|
|
"process",
|
|
should_continue,
|
|
{"retry": "process", "finish": END}
|
|
)
|
|
```
|
|
|
|
### Parallel Execution
|
|
```python
|
|
from langgraph.graph import START
|
|
|
|
# Fan out to parallel nodes
|
|
graph.add_edge(START, ["agent_a", "agent_b", "agent_c"])
|
|
|
|
# Fan in to aggregator
|
|
graph.add_edge(["agent_a", "agent_b", "agent_c"], "synthesize")
|
|
```
|
|
|
|
## Production Deployment
|
|
|
|
### Streaming for UX
|
|
```python
|
|
async for event in app.astream_events(input, version="v2"):
|
|
if event["event"] == "on_chat_model_stream":
|
|
print(event["data"]["chunk"].content, end="")
|
|
```
|
|
|
|
### Error Handling
|
|
```python
|
|
def error_handler(state):
|
|
try:
|
|
return execute_risky_operation(state)
|
|
except Exception as e:
|
|
return {"error": str(e), "next": "fallback"}
|
|
|
|
graph.add_node("risky_op", error_handler)
|
|
graph.add_conditional_edges(
|
|
"risky_op",
|
|
lambda s: "fallback" if "error" in s else "success"
|
|
)
|
|
```
|
|
|
|
### Monitoring with LangSmith
|
|
```python
|
|
import os
|
|
os.environ["LANGCHAIN_TRACING_V2"] = "true"
|
|
os.environ["LANGCHAIN_API_KEY"] = "..."
|
|
|
|
# All agent actions automatically logged to LangSmith
|
|
app.invoke(input)
|
|
```
|
|
|
|
## Best Practices
|
|
|
|
**DO:**
|
|
✅ Use checkpointing for long-running tasks
|
|
✅ Stream outputs for better UX
|
|
✅ Implement human approval for critical actions
|
|
✅ Use conditional edges for complex routing
|
|
✅ Leverage parallel execution when possible
|
|
✅ Monitor with LangSmith in production
|
|
|
|
**DON'T:**
|
|
❌ Use AgentExecutor (deprecated)
|
|
❌ Skip error handling on nodes
|
|
❌ Forget to set thread_id for stateful conversations
|
|
❌ Over-complicate graphs unnecessarily
|
|
❌ Ignore memory management for long conversations
|
|
|
|
## Integration Examples
|
|
|
|
### With Claude
|
|
```python
|
|
from langchain_anthropic import ChatAnthropic
|
|
|
|
llm = ChatAnthropic(model="claude-sonnet-4-5")
|
|
agent = create_react_agent(llm, tools)
|
|
```
|
|
|
|
### With OpenAI
|
|
```python
|
|
from langchain_openai import ChatOpenAI
|
|
|
|
llm = ChatOpenAI(model="gpt-4o")
|
|
agent = create_react_agent(llm, tools)
|
|
```
|
|
|
|
### With MCP Servers
|
|
```python
|
|
from langchain_mcp import MCPTool
|
|
|
|
github_tool = MCPTool.from_server("github-mcp")
|
|
tools = [github_tool, ...]
|
|
agent = create_react_agent(llm, tools)
|
|
```
|
|
|
|
## Decision Framework
|
|
|
|
**Use LangGraph when:**
|
|
- Need fine-grained control over agent execution
|
|
- Building complex state machines
|
|
- Require human-in-the-loop workflows
|
|
- Want production-grade durability (checkpointing)
|
|
- Need to support multiple LLM providers
|
|
|
|
**Use alternatives when:**
|
|
- Want managed platform (use OpenAI AgentKit)
|
|
- Need visual builder (use AgentKit)
|
|
- Want simpler API (use Claude SDK directly)
|
|
- Building on Oracle Cloud only (use Oracle ADK)
|
|
|
|
## Resources
|
|
|
|
- Docs: https://langchain-ai.github.io/langgraph/
|
|
- GitHub: https://github.com/langchain-ai/langgraph
|
|
- Tutorials: https://langchain-ai.github.io/langgraph/tutorials/
|
|
|
|
---
|
|
|
|
*LangGraph is the production-grade choice for complex agentic workflows requiring maximum control.*
|