LangChain Ecosystem Guide
Build production-ready agents with LangGraph, from basic primitives to advanced context management.
Quick Start: Which Tool?
IMPORTANT: Use modern abstractions. Older helpers like create_sql_agent, create_tool_calling_agent, create_react_agent, etc. are outdated.
Simple tool-calling agent? → create_agent
from langchain.agents import create_agent
graph = create_agent(model="anthropic:claude-sonnet-4-5", tools=[search], system_prompt="...")
Use this for: Basic ReAct loops, tool-calling agents, simple Q&A bots.
Need planning + filesystem + subagents? → create_deep_agent
from deepagents import create_deep_agent
agent = create_deep_agent(model=model, tools=tools, backend=FilesystemBackend())
Use this for: Research agents, complex workflows, agents that need to store intermediate results, multi-step planning.
Custom control flow / multi-agent / advanced context? → LangGraph (this guide) Use this for: Custom routing logic, supervisor patterns, specialized state management, non-standard workflows.
Start simple: Build with basic ReAct loops first. Only add complexity (multi-agent, advanced context management) when your use case requires it.
1. Core Primitives
Using create_agent (Recommended Starting Point)
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_agent
from langchain_core.tools import tool
@tool
def my_tool(query: str) -> str:
"""Tool description that the model sees."""
return perform_operation(query)
model = ChatAnthropic(model="claude-sonnet-4-5")
agent = create_agent(
model=model,
tools=[my_tool],
system_prompt="Your agent behavior and guidelines."
)
result = agent.invoke({"messages": [("user", "Your question")]})
Pattern applies to: SQL agents, search agents, Q&A bots, tool-calling workflows.
Replaces: Legacy convenience helpers and LLMChain patterns.
Example: Calculator Agent
from langchain_anthropic import ChatAnthropic
from langchain.agents import create_agent
from langchain_core.tools import tool
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression safely."""
try:
# Only allow safe math operations
allowed = set('0123456789+-*/(). ')
if not all(c in allowed for c in expression):
return "Error: Invalid characters"
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
@tool
def convert_units(value: float, from_unit: str, to_unit: str) -> str:
"""Convert between common units."""
conversions = {
("km", "miles"): 0.621371,
("miles", "km"): 1.60934,
("kg", "lbs"): 2.20462,
("lbs", "kg"): 0.453592,
}
factor = conversions.get((from_unit, to_unit), None)
if factor:
return f"{value * factor:.2f} {to_unit}"
return "Conversion not supported"
model = ChatAnthropic(model="claude-sonnet-4-5")
agent = create_agent(
model=model,
tools=[calculate, convert_units],
system_prompt="You are a helpful calculator assistant."
)
result = agent.invoke({"messages": [("user", "What is 15% of 250?")]})
Pattern applies to: SQL agents, search agents, Q&A bots, tool-calling workflows.
Basic ReAct Agent from Scratch
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
class State(TypedDict):
messages: Annotated[list, add_messages]
tools = [search_tool]
tool_node = ToolNode(tools) # Handles ToolMessage generation
def agent(state: State):
return {"messages": [model.bind_tools(tools).invoke(state["messages"])]}
def route(state: State):
return "tools" if state["messages"][-1].tool_calls else END
# Build graph
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", route)
workflow.add_edge("tools", "agent")
app = workflow.compile()
The loop: Agent → tools → agent → END
ToolMessages: Critical Detail
When implementing custom tool execution, you must create a ToolMessage for each tool call:
def custom_tool_node(state: State) -> dict:
"""Execute tools manually."""
last_message = state["messages"][-1]
tool_messages = []
for tool_call in last_message.tool_calls:
result = execute_tool(tool_call["name"], tool_call["args"])
# CRITICAL: tool_call_id must match!
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"]
))
return {"messages": tool_messages}
Commands: Routing with Updates
from langgraph.types import Command
from typing import Literal
def router(state: State) -> Command[Literal["research", "write", END]]:
"""Route and update state simultaneously."""
if needs_more_context(state):
return Command(
update={"notes": "Starting research phase"},
goto="research"
)
return Command(goto=END)
# Human-in-loop: pause and resume
def ask_user(state: State) -> Command:
response = interrupt("Please clarify:")
return Command(
update={"messages": [HumanMessage(content=response)]},
goto="continue"
)
# Resume: graph.invoke(Command(resume=user_input), config)
2. Context Management Strategies
Strategy 1: Subagent Delegation
Pattern: Offload work to subagents, return only summaries.
# Specialized subagent (compiles full workflow internally)
researcher_subgraph = build_researcher_graph().compile()
# Main agent delegates
def main_agent(state: State) -> Command:
if needs_research(state["messages"][-1]):
result = researcher_subgraph.invoke({"query": extract_query(state)})
# Add ONLY summary to main context
return Command(
update={"context": state["context"] + f"\n{result['summary']}"},
goto="respond"
)
return Command(goto="respond")
Why: Subagent handles complexity, main agent only sees compressed summary.
Strategy 2: Filesystem Context Management
Pattern: Write intermediate work to files, pass file paths instead of full content.
class State(TypedDict):
messages: Annotated[list, add_messages]
context_files: list[str] # File paths, not content
def research_and_save(state: State) -> dict:
research_result = conduct_research(state["messages"][-1])
file_path = Path("workspace") / f"research_{len(state['context_files'])}.json"
with open(file_path, "w") as f:
json.dump(research_result, f)
return {"context_files": [str(file_path)]} # Store path only
def respond_with_context(state: State) -> dict:
# Load only last 3 files
context = [json.load(open(p)) for p in state["context_files"][-3:]]
messages = state["messages"] + [SystemMessage(content=str(context))]
return {"messages": [model.invoke(messages)]}
Why: State stays small, full context lives in files, load selectively.
Strategy 3: Progressive Message Trimming
Pattern: Remove old messages but preserve recent context and critical system messages.
def trim_messages(messages: list, max_messages: int = 20) -> list:
"""Keep system messages + recent conversation."""
# Separate system messages from conversation
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
conversation = [m for m in messages if not isinstance(m, SystemMessage)]
# Keep only recent conversation
recent = conversation[-max_messages:]
# Recombine
return system_msgs + recent
def agent_with_trimming(state: State) -> dict:
"""Call model with trimmed context."""
trimmed = trim_messages(state["messages"], max_messages=15)
response = model.invoke(trimmed)
return {"messages": [response]}
Advanced: Remove by token count
def trim_by_tokens(messages: list, max_tokens: int = 4000) -> list:
"""Trim to fit token budget."""
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
conversation = [m for m in messages if not isinstance(m, SystemMessage)]
# Count tokens (pseudo-code, use tiktoken)
total_tokens = sum(count_tokens(m) for m in conversation)
while total_tokens > max_tokens and len(conversation) > 5:
# Remove oldest non-system message
removed = conversation.pop(0)
total_tokens -= count_tokens(removed)
return system_msgs + conversation
Strategy 4: Compression with Summarization
Pattern: Summarize old context into compact form, keep only recent messages raw.
def compress_history(state: State) -> dict:
"""Compress old messages into summary."""
messages = state["messages"]
# If conversation too long, compress
if len(messages) > 30:
# Separate into old (to compress) and recent (keep raw)
old_messages = messages[:-10]
recent_messages = messages[-10:]
# Summarize old messages
summary_prompt = f"""Summarize this conversation history concisely:
{format_messages(old_messages)}"""
summary = model.invoke([HumanMessage(content=summary_prompt)])
# Replace old messages with summary
compressed = [
SystemMessage(content=f"Previous context:\n{summary.content}")
] + recent_messages
return {"messages": {"type": "override", "value": compressed}}
return {} # No compression needed
# Custom reducer for override
def override_reducer(current: list, update: dict | list) -> list:
if isinstance(update, dict) and update.get("type") == "override":
return update["value"]
return current + (update if isinstance(update, list) else [update])
class State(TypedDict):
messages: Annotated[list, override_reducer]
Strategy 5: Hierarchical Context with State Layers
Pattern: Separate working memory (ephemeral) from long-term context (persistent).
class WorkingState(TypedDict):
"""Ephemeral working memory."""
messages: Annotated[list, add_messages]
current_task: str
class LongTermState(TypedDict):
"""Persistent context across tasks."""
user_id: str
learned_facts: list[str]
conversation_summaries: list[str]
class FullState(WorkingState, LongTermState):
"""Combined state."""
pass
def agent_with_context_layers(state: FullState, store: BaseStore) -> dict:
"""Use hierarchical context."""
# Load long-term context
namespace = ("users", state["user_id"])
learned = store.get(namespace, "facts")
# Build prompt with layers
long_term_context = "\n".join(learned.value if learned else [])
working_context = state["current_task"]
prompt = f"""Long-term context:
{long_term_context}
Current task: {working_context}
Recent conversation:
{format_messages(state["messages"][-5:])}"""
response = model.invoke([SystemMessage(content=prompt)])
return {"messages": [response]}
def save_learnings(state: FullState, store: BaseStore) -> dict:
"""Persist important facts."""
namespace = ("users", state["user_id"])
# Extract learnings from conversation
learnings = extract_key_facts(state["messages"])
# Append to long-term storage
existing = store.get(namespace, "facts")
all_facts = (existing.value if existing else []) + learnings
store.put(namespace, "facts", all_facts)
return {}
Strategy 6: Dynamic Subagent Context Isolation
Pattern: Spawn subagents with only relevant context slice, not entire conversation.
def spawn_focused_subagent(state: State, task: str) -> dict:
"""Create subagent with minimal context."""
# Extract only relevant messages for this task
relevant_msgs = filter_relevant_messages(state["messages"], task)
# Create subagent with focused context
subagent_state = {
"messages": relevant_msgs[-5:], # Only last 5 relevant msgs
"task": task
}
result = subagent.invoke(subagent_state)
# Bring back only the final answer
return {
"messages": [AIMessage(content=f"Task '{task}' complete: {result['answer']}")]
}
def filter_relevant_messages(messages: list, task: str) -> list:
"""Use embedding similarity to find relevant messages."""
task_embedding = embed(task)
scored = []
for msg in messages:
if isinstance(msg, (HumanMessage, AIMessage)):
similarity = cosine_similarity(task_embedding, embed(msg.content))
scored.append((similarity, msg))
# Return top-k most relevant
scored.sort(reverse=True)
return [msg for _, msg in scored[:10]]
3. Multi-Agent Patterns
Supervisor with Context Budget
class SupervisorState(TypedDict):
messages: Annotated[list, add_messages]
research_notes: list[str] # Summaries only
next_agent: str
def supervisor(state: SupervisorState) -> Command[Literal["researcher", "writer", END]]:
"""Coordinate agents with compressed context."""
# Build minimal context for routing decision
recent = state["messages"][-3:]
summaries = "\n".join(state["research_notes"][-2:]) # Only recent summaries
routing_prompt = f"""Recent conversation: {recent}
Research summaries: {summaries}
What should we do next?"""
decision = routing_model.with_structured_output(Routing).invoke([
HumanMessage(content=routing_prompt)
])
if decision.done:
return Command(goto=END)
return Command(goto=decision.next_agent)
def researcher(state: SupervisorState) -> dict:
"""Research and return summary only."""
research_result = conduct_research(state["messages"][-1])
# Compress research into summary
summary = summarize(research_result)
return {
"research_notes": [summary], # Add summary, not full result
"messages": [AIMessage(content="Research complete")]
}
Parallel Subagents with Result Synthesis
async def parallel_research_with_synthesis(state: State) -> dict:
"""Run multiple subagents, synthesize results."""
query = state["messages"][-1].content
# Spawn 3 researchers with different strategies
tasks = [
researcher_1.ainvoke({"query": query, "strategy": "academic"}),
researcher_2.ainvoke({"query": query, "strategy": "news"}),
researcher_3.ainvoke({"query": query, "strategy": "social"})
]
results = await asyncio.gather(*tasks)
# Synthesize into single summary
synthesis_prompt = f"""Synthesize these research results:
1. Academic: {results[0]['summary']}
2. News: {results[1]['summary']}
3. Social: {results[2]['summary']}
Provide unified 3-sentence summary:"""
synthesis = model.invoke([HumanMessage(content=synthesis_prompt)])
# Return only synthesis, not individual results
return {"messages": [synthesis]}
4. Practical Patterns
Checkpointer + Store for Persistence
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
checkpointer = MemorySaver() # Thread-level state
store = InMemoryStore() # Cross-thread memory
app = graph.compile(
checkpointer=checkpointer, # Enables conversation continuity
store=store # Enables learning across conversations
)
# Thread-specific conversation
app.invoke(
{"messages": [HumanMessage("Hello")]},
config={"configurable": {"thread_id": "user-123"}}
)
Structured Output for Reliability
from pydantic import BaseModel, Field
class ResearchOutput(BaseModel):
summary: str = Field(description="3-sentence summary")
sources: list[str] = Field(description="Source URLs")
confidence: float = Field(description="0-1 confidence score")
model_with_structure = model.with_structured_output(ResearchOutput)
def structured_research(state: State) -> dict:
result = model_with_structure.invoke(state["messages"])
# result is guaranteed to have summary, sources, confidence
return {"research": result.model_dump()}
Middleware for Cross-Cutting Concerns
Pattern: Wrap nodes with logging, timing, error handling, or filtering.
def logging_middleware(func):
"""Log inputs/outputs for any node."""
def wrapper(state):
print(f"[{func.__name__}] Input: {state}")
result = func(state)
print(f"[{func.__name__}] Output: {result}")
return result
return wrapper
# Apply to nodes
workflow.add_node("agent", logging_middleware(agent))
# Or apply at compile time for all nodes
app = workflow.compile(middleware=[logging_middleware])
5. DeepAgents: Batteries Included
When you need context management built-in:
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, FilesystemBackend, StoreBackend
# Mix filesystem (working files) + store (memories)
backend = CompositeBackend({
"/workspace/": FilesystemBackend("./workspace"),
"/memories/": StoreBackend(store)
})
agent = create_deep_agent(
model=model,
tools=[search, scrape],
subagents=[researcher_agent, analyst_agent],
backend=backend
)
DeepAgents provides:
- Filesystem: Write/read context files automatically
- Planning: Break tasks into steps
- Subagents: Delegate to specialists
- Memory: Persistent storage across sessions
Quick Reference
Context Management:
# Subagent delegation → compress result
result = subagent.invoke(query)
return {"context": result["summary"]}
# Filesystem → store path, not content
with open(file_path, "w") as f: json.dump(data, f)
return {"files": [str(file_path)]}
# Trim messages → keep recent
messages[-20:]
# Compress → summarize old
summary = model.invoke(f"Summarize: {old_msgs}")
return {"messages": [SystemMessage(summary)] + recent}
# Hierarchical → separate working/long-term
prompt = f"{long_term_context}\n{working_context}"
# Focused subagent → relevant slice only
subagent.invoke({"messages": relevant[-5:]})
Commands:
Command(update={...}, goto="next")
Command(goto=END)
Command(resume=user_input)
Tool Messages:
ToolMessage(content=result, tool_call_id=call["id"])
State Override:
{"messages": {"type": "override", "value": compressed}}