Haystack Tutorial (Python): persisting agent state for beginners

This tutorial shows you how to persist an agent’s state in Haystack so a conversation or task can survive process restarts. You need this when you want memory across requests, durable tool history, or a chat app that does not lose context every time the server restarts.

What You'll Need

• •Python 3.10+
• •haystack-ai installed
• •An OpenAI API key
• •A working internet connection for model calls
• •Basic familiarity with Haystack Pipeline, ChatMessage, and components

Install the package:

pip install haystack-ai

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start by creating a small chat pipeline that uses a persistent memory store. The key idea is to keep messages outside the process so you can reload them later.

from haystack import Pipeline, component
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator

@component
class MemoryWriter:
    def __init__(self):
        self.messages = []

    @component.output_types(messages=list[ChatMessage])
    def run(self, messages: list[ChatMessage]):
        self.messages.extend(messages)
        return {"messages": self.messages}

pipeline = Pipeline()
pipeline.add_component("memory", MemoryWriter())
pipeline.add_component("chat", OpenAIChatGenerator(model="gpt-4o-mini"))
pipeline.connect("memory.messages", "chat.messages")

2. •Next, define a tiny persistence layer. For beginners, JSON is enough to prove the pattern before moving to Redis or a database.

import json
from pathlib import Path

STATE_FILE = Path("agent_state.json")

def save_state(messages):
    data = [{"role": m.role.value, "content": m.content} for m in messages]
    STATE_FILE.write_text(json.dumps(data, indent=2))

def load_state():
    if not STATE_FILE.exists():
        return []
    raw = json.loads(STATE_FILE.read_text())
    return [ChatMessage.from_dict(item) for item in raw]

3. •Now wire the persisted state into the pipeline run. You load prior messages, append the new user message, run the model, then save the updated transcript back to disk.

history = load_state()

user_message = ChatMessage.from_user("My name is Sam. Remember it.")
history.append(user_message)

result = pipeline.run(
    {
        "memory": {"messages": history},
        "chat": {"messages": history},
    }
)

assistant_message = result["chat"]["replies"][0]
history.append(assistant_message)
save_state(history)

print(assistant_message.content)

4. •To make this useful across multiple turns, repeat the same load-run-save flow each time you get a new user input. The persisted file becomes your agent state, and every request rebuilds context from disk before calling the model.

def chat_once(user_text: str):
    history = load_state()
    history.append(ChatMessage.from_user(user_text))

    result = pipeline.run(
        {
            "memory": {"messages": history},
            "chat": {"messages": history},
        }
    )

    reply = result["chat"]["replies"][0]
    history.append(reply)
    save_state(history)
    return reply.content

print(chat_once("What is my name?"))
print(chat_once("And what did I ask you to remember?"))

5. •If you want cleaner production behavior, separate state from inference logic. The pattern stays the same: hydrate state, run the model, persist the updated state after success.

def handle_request(user_text: str) -> str:
    history = load_state()
    history.append(ChatMessage.from_user(user_text))

    result = pipeline.run(
        {
            "memory": {"messages": history},
            "chat": {"messages": history},
        }
    )

    reply = result["chat"]["replies"][0]
    history.append(reply)
    save_state(history)
    return reply.content

if __name__ == "__main__":
    print(handle_request("Store that I work in insurance claims."))

Testing It

Run the script once and send a message like “My name is Sam.” Then restart Python and ask “What is my name?” If persistence is working, the second run should still have access to the first message because it was reloaded from agent_state.json.

Open the JSON file and confirm both user and assistant messages are being appended after each turn. If you only see one message or an empty file, your save step is failing.

For a stricter test, delete the file and rerun the script. The agent should behave like a fresh session on first launch and like a remembered session on subsequent launches.

Next Steps

• •Replace JSON with Redis if you need concurrent users or multiple app instances.
• •Add message trimming so old context does not grow without bound.
• •Move from plain chat memory to structured agent state with tool results and task metadata

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.