Agent State and Memory

Capsule overview

Agents built with create_agent carry state — an object that accumulates the conversation, tool results, and any custom data you define. Understanding how state works is the key to building agents that remember context, gather information, and make decisions based on everything that has happened during the session.

In this capsule you'll learn how the default state works (the messages list that grows automatically), how to extend it with custom fields using TypedDict, how to reach the state from dynamic prompts and tools, and the difference between state (mutable per run) and config (immutable). By the end, you'll be able to design the state for any agent your application needs.

Everything here builds on the previous capsule — there you saw how dynamic prompts read state. Here you'll learn how to design what that state holds.

The default state: messages

When you create an agent with create_agent, the default state includes a messages list that grows automatically with each turn of the ReAct loop:

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_price(product: str) -> str:
    """Get the price of a product."""
    prices = {"laptop": "$999", "mouse": "$29", "keyboard": "$79"}
    return prices.get(product.lower(), f"Product '{product}' not found")

model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [get_price])

result = agent.invoke({
    "messages": [("user", "How much does a mouse cost?")]
})

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content[:100] if msg.content else '[tool_call]'}")
# Expected output:
# human: How much does a mouse cost?
# ai: [tool_call for get_price]
# tool: $29
# ai: The mouse costs $29.

The messages list holds the entire history: user messages (HumanMessage), model responses (AIMessage, with tool calls or text), and tool results (ToolMessage). Every turn of the ReAct loop appends more messages until the agent decides to answer without calling tools.

Custom state with state_schema

Why extend the state

The default state only has messages. But real applications need to track more:

• ✅ Sources found during research
• ✅ Counters (attempts, searches performed)
• ✅ Control flags (research complete, user verified)
• ✅ Accumulated data (URLs visited, products compared)

To add custom fields, you use state_schema with a TypedDict:

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sources_found: list[str]
    research_complete: bool

@tool
def search_web(query: str) -> str:
    """Search the internet for information."""
    return f"Result for '{query}': [information found]"

model = ChatOpenAI(model="gpt-4.1-mini")

agent = create_agent(model, [search_web], state_schema=ResearchState)

result = agent.invoke({
    "messages": [("user", "Research LangChain")],
    "sources_found": [],
    "research_complete": False,
})
print(result["messages"][-1].content)
# Expected output: [The agent's answer about LangChain]

Understanding Annotated and operator.add

The messages field always needs Annotated[list[AnyMessage], operator.add]. This tells LangGraph how to merge new values with existing ones:

from typing import TypedDict, Annotated
from langchain_core.messages import AnyMessage
import operator

class MyState(TypedDict):
    # operator.add = new messages get APPENDED to the existing list
    messages: Annotated[list[AnyMessage], operator.add]

    # No Annotated = the value gets REPLACED entirely
    current_topic: str
    search_count: int

The rule is simple:

• ✅ Annotated[list, operator.add] → values accumulate (append)
• ✅ No Annotated → the value gets replaced entirely

For messages, you always use operator.add because you want the history to grow. For fields like counters or flags, you use plain types without Annotated because you want to overwrite them.

Initializing the state

When you invoke an agent with a custom state_schema, you pass the initial values for every field:

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class ShoppingState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    budget: float
    items_in_cart: list[str]

@tool
def search_product(name: str) -> str:
    """Search for a product by name."""
    catalog = {"laptop": "Laptop Pro — $999.99", "mouse": "Mouse Wireless — $29.99", "monitor": "Monitor 27\" — $349.99"}
    return catalog.get(name.lower(), f"Product '{name}' not found")

model = ChatOpenAI(model="gpt-4.1-mini")

agent = create_agent(
    model, [search_product],
    state_schema=ShoppingState,
    prompt="You are a shopping assistant. The user has a limited budget. Help them find products within it. Respond in English.",
)

result = agent.invoke({
    "messages": [("user", "I'm looking for a mouse")],
    "budget": 100.00,
    "items_in_cart": [],
})
print(result["messages"][-1].content)
# Expected output: [Answer with the mouse it found and its price]

If you leave out a field that's in state_schema, its value will be None. It's good practice to always pass every field explicitly.

Reading the state from dynamic prompts

In the previous capsule you saw dynamic prompts. Now that you know how to define custom state, you can combine the two to build highly personalized agents:

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class TutorState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_name: str
    user_level: str
    queries_remaining: int

@tool
def search_docs(query: str) -> str:
    """Search the technical documentation."""
    return f"Documentation for '{query}': [results]"

model = ChatOpenAI(model="gpt-4.1-mini")

def personalized_prompt(state):
    """Build the prompt from the agent's full state."""
    name = state.get("user_name", "user")
    level = state.get("user_level", "beginner")
    remaining = state.get("queries_remaining", 10)

    level_instructions = {
        "beginner": "Explain concepts simply, with analogies.",
        "intermediate": "Give direct technical explanations with code.",
        "advanced": "Be concise and technical. Focus on edge cases.",
    }
    instructions = level_instructions.get(level, level_instructions["beginner"])

    prompt = f"You are a programming tutor helping {name}. Level: {level}. {instructions} Queries remaining: {remaining}."
    if remaining <= 2:
        prompt += " Let the user know they have few queries left."
    prompt += " Respond in English."
    return prompt

agent = create_agent(model, [search_docs], prompt=personalized_prompt, state_schema=TutorState)

result = agent.invoke({
    "messages": [("user", "How does async/await work?")],
    "user_name": "Laura",
    "user_level": "intermediate",
    "queries_remaining": 5,
})
print(result["messages"][-1].content)
# Expected output: [A technical explanation of async/await pitched at intermediate level]

Reading the state from tools

Tools can receive immutable information via config using RunnableConfig:

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig
import operator

class AppState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_id: str

@tool
def get_user_data(field: str, config: RunnableConfig) -> str:
    """Get data about the current user. field: 'profile', 'orders', 'settings'."""
    user_id = config.get("configurable", {}).get("user_id", "unknown")
    user_db = {"USR-001": {"profile": "María García — Premium", "orders": "3 active orders", "settings": "Language: EN"}}
    return user_db.get(user_id, {}).get(field, f"Field '{field}' not found for {user_id}")

model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [get_user_data], state_schema=AppState, prompt="You are an account assistant. Respond in English.")

result = agent.invoke(
    {"messages": [("user", "What are my orders?")], "user_id": "USR-001"},
    config={"configurable": {"user_id": "USR-001"}},
)
print(result["messages"][-1].content)
# Expected output: [Answer with the user's 3 active orders]

The config gets passed as the second argument to invoke(), and the tool receives it through the config: RunnableConfig parameter.

Conversation history: short-term memory

The messages state acts as short-term memory — the agent remembers everything that happened in the current session. You can simulate multi-turn conversations by passing the accumulated history:

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [calculate], prompt="You are a math tutor. Remember the conversation's context. Respond in English.")

conversation = []
questions = [
    "What's 15 * 23?",
    "Now divide that result by 5",
    "And what if I add 100?"
]

for question in questions:
    conversation.append(("user", question))
    result = agent.invoke({"messages": conversation.copy()})

    response = result["messages"][-1]
    conversation = [(msg.type if hasattr(msg, 'type') else msg[0],
                      msg.content if hasattr(msg, 'content') else msg[1])
                     for msg in result["messages"]]

    print(f"User: {question}")
    print(f"Agent: {response.content}\n")
# Expected output:
# User: What's 15 * 23?
# Agent: 15 × 23 = 345.
#
# User: Now divide that result by 5
# Agent: 345 ÷ 5 = 69.
#
# User: And what if I add 100?
# Agent: 69 + 100 = 169.

Each turn, the agent receives the whole history — which is why it resolves "that result" without ambiguity. For persistence across sessions, you need checkpointing (Module 8).

State vs Config: what goes where

result = agent.invoke(
    {
        # State: data that changes during the run
        "messages": [("user", "Analyze the sales")],
        "queries_executed": [],
        "analysis_complete": False,
    },
    # Config: immutable data
    config={"configurable": {"database": "production", "user_id": "analyst-42"}},
)

Rule of thumb:

• ✅ If the data changes during the run → state
• ✅ If the data is constant for the whole run → config

Designing the state: a full example

When you design the state, combine the different field types your agent needs:

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class ResearchAgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    researcher_name: str
    research_topic: str
    sources: list[str]
    key_findings: list[str]
    searches_performed: int
    max_searches: int
    research_complete: bool

@tool
def web_search(query: str) -> str:
    """Search the internet for information on a topic."""
    return f"Results for '{query}': [3 relevant articles found]"

@tool
def summarize_findings(findings: str) -> str:
    """Write a summary of the research findings."""
    return f"Summary: {findings[:200]}..."

model = ChatOpenAI(model="gpt-4.1")

def research_prompt(state):
    name = state.get("researcher_name", "researcher")
    topic = state.get("research_topic", "undefined topic")
    searches = state.get("searches_performed", 0)
    max_s = state.get("max_searches", 5)

    if state.get("research_complete", False):
        return f"The research on '{topic}' is complete. Summarize the findings. Respond in English."

    remaining = max_s - searches
    return (
        f"You are a research assistant helping {name}. "
        f"Topic: {topic}. Searches: {searches}/{max_s} ({remaining} left). "
        f"Look for relevant information. If you already have enough or you're out of searches, write the final summary. Respond in English."
    )

agent = create_agent(model, [web_search, summarize_findings], prompt=research_prompt, state_schema=ResearchAgentState)

result = agent.invoke({
    "messages": [("user", "Research the current trends in generative AI")],
    "researcher_name": "Carlos",
    "research_topic": "generative AI 2026",
    "sources": [],
    "key_findings": [],
    "searches_performed": 0,
    "max_searches": 3,
    "research_complete": False,
})
print(result["messages"][-1].content)
# Expected output: [The agent's answer with the research results]

Checklist: (1) What should it remember between turns? → messages. (2) Does it accumulate info? → list[str]. (3) Counters? → int. (4) Flags? → bool. (5) Context for the prompt? → str.

Connection to the project

In the Research Agent with Tools (this module's project):

• You'll define a ResearchState with fields for sources, findings, and control flags
• The dynamic prompt will read the state to adapt behavior to the progress so far
• The agent will accumulate sources and findings across multiple tool calls
• In Capsule 05 (Streaming), you'll see how to watch the state evolve in real time

Everything you learn here applies directly in Capsule 08.

Troubleshooting

Problem 1: "KeyError" when reading state fields

Cause: The field wasn't passed to invoke(), or it isn't defined in state_schema. Fix: Always pass every state_schema field in invoke(), and use state.get("field", default) in dynamic prompts to handle missing values.

Problem 2: Messages get lost between invocations

Cause: Every invoke() is an independent run. State doesn't persist across calls. Fix: Accumulate the messages manually between calls:

history = []

# Turn 1
history.append(("user", "What is Python?"))
result = agent.invoke({"messages": history.copy()})
history = [(m.type, m.content) for m in result["messages"]]

# Turn 2 — includes the history from turn 1
history.append(("user", "And what is it used for?"))
result = agent.invoke({"messages": history.copy()})

For automatic persistence across sessions, you need checkpointing (Module 8).

Problem 3: state_schema won't accept the fields I pass

Cause: The types in invoke() don't match the ones declared in the TypedDict. Fix: Check that the types line up: if you declare count: int, pass count=5 (not count="5").

Problem 4: The messages list grows without limit

Cause: The history accumulates every message (including intermediate tool calls) and can blow past the context window. Fix:

def trim_messages(messages, max_messages=20):
    """Keep only the last N messages."""
    if len(messages) <= max_messages:
        return messages
    return messages[-max_messages:]

trimmed = trim_messages(history, max_messages=20)
result = agent.invoke({"messages": trimmed})

For advanced strategies (summarization, sliding window), see Module 8.

Exercises

Exercise 1: State with a search counter (Easy)

Define a state_schema with search_count and max_searches. Write a dynamic prompt that warns the user when fewer than 2 searches remain.

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class SearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    search_count: int
    max_searches: int

@tool
def search(query: str) -> str:
    """Search the internet for information."""
    return f"Results for '{query}': [data found]"

model = ChatOpenAI(model="gpt-4.1-mini")

def search_aware_prompt(state):
    count = state.get("search_count", 0)
    max_s = state.get("max_searches", 5)
    remaining = max_s - count

    prompt = f"You are a research assistant. Searches: {count}/{max_s}. "
    if remaining <= 2:
        prompt += f"⚠️ Only {remaining} searches left. Prioritize direct questions. Warn the user. "
    else:
        prompt += "Search freely to find the best information. "
    prompt += "Respond in English."
    return prompt

agent = create_agent(model, [search], prompt=search_aware_prompt, state_schema=SearchState)

result = agent.invoke({
    "messages": [("user", "Look up information about LangChain")],
    "search_count": 3,
    "max_searches": 5,
})
print(result["messages"][-1].content)
# Expected output: [Answer with a warning that few searches remain]

Exercise 2: State with an accumulated source list (Easy)

Define a state with sources: list[str]. The dynamic prompt should list the sources already consulted so the agent doesn't repeat searches.

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class SourceState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sources: list[str]

@tool
def search_article(topic: str) -> str:
    """Search for articles on a specific topic."""
    articles = {
        "langchain": "Article: 'LangChain v1.2 Release Notes' (langchain.com)",
        "agents": "Article: 'Building AI Agents in 2026' (arxiv.org)",
    }
    for key, value in articles.items():
        if key in topic.lower():
            return value
    return f"No articles found on '{topic}'"

model = ChatOpenAI(model="gpt-4.1-mini")

def source_aware_prompt(state):
    sources = state.get("sources", [])
    prompt = "You are a researcher gathering technical articles. "
    if sources:
        prompt += f"You've already consulted: [{', '.join(sources)}]. Don't repeat searches on the same topics. "
    else:
        prompt += "You haven't consulted any sources yet. Start the research. "
    prompt += "Respond in English."
    return prompt

agent = create_agent(model, [search_article], prompt=source_aware_prompt, state_schema=SourceState)

result = agent.invoke({
    "messages": [("user", "Research agents in AI")],
    "sources": ["langchain.com"],
})
print(result["messages"][-1].content)
# Expected output: [Answer citing new sources, without repeating langchain.com]

Exercise 3: Complex state for e-commerce (Medium)

Design a state_schema with a cart (list), a budget (float), and a pending checkout (bool). The dynamic prompt should tailor recommendations to the remaining budget.

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator

class EcommerceState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    cart: list[dict]
    budget: float
    checkout_pending: bool

@tool
def search_products(category: str) -> str:
    """Search for products by category."""
    catalog = {
        "electronics": "[BT Headphones $49.99, USB-C Charger $19.99, HD Webcam $89.99]",
        "books": "[Clean Code $35.00, Design Patterns $42.00]",
    }
    return catalog.get(category.lower(), f"No products in '{category}'")

model = ChatOpenAI(model="gpt-4.1")

def shopping_prompt(state):
    cart = state.get("cart", [])
    budget = state.get("budget", 0)
    total_in_cart = sum(item.get("price", 0) for item in cart)
    remaining = budget - total_in_cart

    if state.get("checkout_pending", False):
        return f"The user is ready to pay. Cart: {len(cart)} products for ${total_in_cart:.2f}. Confirm and proceed. Respond in English."

    prompt = f"You are a shopping assistant. Budget: ${budget:.2f}. In cart: {len(cart)} (${total_in_cart:.2f}). Remaining: ${remaining:.2f}. "
    if remaining < 20:
        prompt += "The budget is nearly gone — only suggest cheap products, or ask whether they want to check out. "
    prompt += "Respond in English."
    return prompt

agent = create_agent(model, [search_products], prompt=shopping_prompt, state_schema=EcommerceState)

result = agent.invoke({
    "messages": [("user", "What electronics do you have?")],
    "cart": [{"name": "Clean Code", "price": 35.00}],
    "budget": 100.00,
    "checkout_pending": False,
})
print(result["messages"][-1].content)
# Expected output: [Products suggested within the remaining $65 budget]

Exercise 4: Multi-turn conversation with history (Medium)

Implement a 3-turn loop where the agent remembers context: ask about a topic, ask for detail, then ask for a summary.

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def search_topic(query: str) -> str:
    """Search for information on a topic."""
    results = {
        "langchain": "LangChain is a framework for LLM applications. v1.2+ includes create_agent.",
        "agents": "Agents use the ReAct pattern: they reason and act by calling tools.",
    }
    for key, value in results.items():
        if key in query.lower():
            return value
    return f"Information on '{query}': [general data]"

model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [search_topic], prompt="You are a LangChain tutor. Remember the conversation. Respond in English.")

conversation_history = []
turns = ["What is LangChain?", "Tell me more about agents", "Summarize all of it in 3 bullet points"]

for question in turns:
    conversation_history.append(("user", question))
    result = agent.invoke({"messages": conversation_history.copy()})
    response = result["messages"][-1]

    conversation_history = []
    for msg in result["messages"]:
        if hasattr(msg, 'type') and hasattr(msg, 'content'):
            conversation_history.append((msg.type, msg.content if msg.content else "[tool_call]"))

    print(f"User: {question}")
    print(f"Agent: {response.content[:120]}...\n")
# Expected output:
# User: What is LangChain?
# Agent: [Explanation of LangChain]...
# User: Tell me more about agents
# Agent: [Detail on agents, building on the previous turn]...
# User: Summarize all of it in 3 bullet points
# Agent: [Summary of the whole conversation]...

Exercise 5: Telling state from config in a real agent (Hard)

Build an analytics agent where state holds mutable data (queries run, results) and config holds immutable data (database, user_id). The tool reads from config, the prompt from state.

from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig
import operator

class AnalyticsState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    queries_executed: list[str]
    results_summary: list[str]
    analysis_complete: bool

@tool
def run_query(sql: str, config: RunnableConfig) -> str:
    """Run a SQL query against the configured database."""
    db = config.get("configurable", {}).get("database", "unknown")
    mock = {"SELECT COUNT(*) FROM users": f"[{db}] 15,234 users", "SELECT AVG(revenue) FROM sales": f"[{db}] $42.50 average"}
    for pattern, result in mock.items():
        if pattern.lower() in sql.lower():
            return result
    return f"[{db}] Query: {sql} — 0 results"

model = ChatOpenAI(model="gpt-4.1")

def analytics_prompt(state):
    queries = state.get("queries_executed", [])
    if state.get("analysis_complete", False):
        return "Analysis complete. Write the final report. Respond in English."
    prompt = f"You are a data analyst. Queries run: {len(queries)}. "
    prompt += "Run queries to answer the question. Respond in English."
    return prompt

agent = create_agent(model, [run_query], prompt=analytics_prompt, state_schema=AnalyticsState)

result = agent.invoke(
    {"messages": [("user", "How many users do we have?")], "queries_executed": [], "results_summary": [], "analysis_complete": False},
    config={"configurable": {"database": "production_analytics", "user_id": "analyst-maria"}},
)
print(result["messages"][-1].content)
# Expected output: [Report with the total users from production_analytics]

Summary

In this capsule you learned:

• The default create_agent state includes a messages list that accumulates automatically
• You can extend the state with state_schema using TypedDict to add custom fields
• Annotated[list, operator.add] makes values accumulate; without Annotated, they get replaced
• Typical custom fields are: counters, accumulating lists, control flags, and user data
• Dynamic prompts reach the state via state.get("field")
• Tools reach immutable data via config: RunnableConfig
• State = data that changes during the run; Config = immutable data
• The messages history acts as short-term memory within the session
• For persistence across sessions, you need checkpointing (Module 8)

Next capsule: Streaming Agents — you'll learn to watch the agent's reasoning process in real time with agent.stream().

Further reading

1. State Management in LangGraph — Conceptual guide to state
2. How to create agents — Official create_agent reference
3. TypedDict (Python docs) — Official TypedDict documentation
4. Annotated Types / Reducers — How reducers work
5. RunnableConfig — config documentation
6. Messages in LangChain — Message types and accumulation
7. Memory Concepts — Short-term vs long-term memory
8. Trim Messages — Handling long histories

Module 3 — LangChain & LangGraph: From Chains to Agents