langchain-ai 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, multi-step planning.
Custom control flow / multi-agent / advanced context? → LangGraph (see below) 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 when your use case requires it.
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.
Example: Calculator Agent
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression safely."""
try:
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,
}
factor = conversions.get((from_unit, to_unit), None)
return f"{value * factor:.2f} {to_unit}" if factor else "Conversion not supported"
agent = create_agent(
model=ChatAnthropic(model="claude-sonnet-4-5"),
tools=[calculate, convert_units],
system_prompt="You are a helpful calculator assistant."
)
Quick Reference
from langchain.agents import create_agent
agent = create_agent(model=model, tools=[my_tool], system_prompt="...")
result = agent.invoke({"messages": [("user", "question")]})
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
tools = [search_tool]
tool_node = ToolNode(tools)
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
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:
from langchain_core.messages import ToolMessage
def custom_tool_node(state: State) -> dict:
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]]:
if needs_more_context(state):
return Command(update={"notes": "Starting research"}, goto="research")
return Command(goto=END)
# Human-in-loop
def ask_user(state: State) -> Command:
response = interrupt("Please clarify:")
return Command(update={"messages": [HumanMessage(content=response)]}, goto="continue")
Pattern: Offload work to subagents, return only summaries.
researcher_subgraph = build_researcher_graph().compile()
def main_agent(state: State) -> Command:
if needs_research(state["messages"][-1]):
result = researcher_subgraph.invoke({"query": extract_query(state)})
return Command(
update={"context": state["context"] + f"\n{result['summary']}"},
goto="respond"
)
return Command(goto="respond")
Strategy 2: Progressive Message Trimming
Pattern: Remove old messages but preserve system messages and recent context.
def trim_messages(messages: list, max_messages: int = 20) -> list:
system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
conversation = [m for m in messages if not isinstance(m, SystemMessage)]
return system_msgs + conversation[-max_messages:]
def agent_with_trimming(state: State) -> dict:
trimmed = trim_messages(state["messages"], max_messages=15)
return {"messages": [model.invoke(trimmed)]}
Strategy 3: Compression with Summarization
Pattern: Summarize old context, keep recent messages raw.
def compress_history(state: State) -> dict:
messages = state["messages"]
if len(messages) > 30:
old, recent = messages[:-10], messages[-10:]
summary = model.invoke([HumanMessage(content=f"Summarize:\n{format_messages(old)}")])
return {"messages": [SystemMessage(content=f"Previous:\n{summary.content}")] + recent}
return {"messages": messages}
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
next_agent: str
def supervisor(state: AgentState) -> Command[Literal["billing", "technical", END]]:
last_msg = state["messages"][-1].content.lower()
if "invoice" in last_msg or "payment" in last_msg:
return Command(goto="billing")
elif "error" in last_msg or "not working" in last_msg:
return Command(goto="technical")
return Command(goto=END)
def billing_agent(state: AgentState) -> dict:
return {"messages": [billing_model.invoke(state["messages"])]}
def technical_agent(state: AgentState) -> dict:
return {"messages": [tech_model.invoke(state["messages"])]}
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("billing", billing_agent)
workflow.add_node("technical", technical_agent)
workflow.add_edge(START, "supervisor")
workflow.add_edge("billing", END)
workflow.add_edge("technical", END)
app = workflow.compile()
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, store=store)
app.invoke(
{"messages": [HumanMessage("Hello")]},
config={"configurable": {"thread_id": "user-123"}}
)
Structured Output
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"])
return {"research": result.model_dump()}
DeepAgents: Batteries Included
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, FilesystemBackend, StoreBackend
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 (auto context files), Planning (task breakdown), Subagents (delegation), Memory (persistence).