langchain-agents

What this skill does

# 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`](https://docs.langchain.com/oss/python/langchain/agents)
```python
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`](https://docs.langchain.com/oss/python/deepagents/overview)
```python
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)

```python
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

```python
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

```python
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:

```python
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

```python
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.

```python
# 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.

```python
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.

```python
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["