### Install Latest Experimental Features

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Install the latest version of the experimental package to access all available experiments. This command ensures you have the most up-to-date features.

```sh
pip install -U haystack-experimental
```

--------------------------------

### Install Temporary Dependencies with uv pip

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Install packages temporarily for experimentation. This command should only be used for short-term needs and not for permanent dependency management.

```bash
uv pip install PACKAGE
```

--------------------------------

### Install from Main Branch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Install the experimental package directly from the main branch of the GitHub repository to try the very newest features. Note that compatibility with older Haystack versions is not guaranteed.

```sh
pip install git+https://github.com/deepset-ai/haystack-experimental.git@main
```

--------------------------------

### Run Scripts with Test Dependencies using Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Execute Python scripts within an environment that includes test dependencies. Ensure Hatch is installed and configured.

```bash
hatch -e test run python SCRIPT.py
```

--------------------------------

### Agent with HITL and Memory Support

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Demonstrates setting up an Agent with persistent chat history using InMemoryChatMessageStore and configuring a BlockingConfirmationStrategy for a specific tool. This allows for human review of tool calls before execution and maintains conversation context.

```python
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.human_in_the_loop import AlwaysAskPolicy, BlockingConfirmationStrategy, RichConsoleUI
from haystack.tools import create_tool_from_function

from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore
from haystack_experimental.components.agents import Agent

# Define a simple tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Sunny, 22°C in {city}"

weather_tool = create_tool_from_function(get_weather)

# Persistent chat memory store
store = InMemoryChatMessageStore(last_k=20)

agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o"),
    tools=[weather_tool],
    system_prompt="You are a helpful weather assistant.",
    max_agent_steps=10,
    confirmation_strategies={
        weather_tool.name: BlockingConfirmationStrategy(
            confirmation_policy=AlwaysAskPolicy(),
            confirmation_ui=RichConsoleUI(),
        )
    },
    chat_message_store=store,
)
agent.warm_up()

# First turn
result = agent.run(
    messages=[ChatMessage.from_user("What's the weather in Berlin?")],
    chat_message_store_kwargs={"chat_history_id": "user_1"},
)
print(result["last_message"].text)

# Second turn — history is automatically retrieved from the store
result2 = agent.run(
    messages=[ChatMessage.from_user("And in Tokyo?")],
    chat_message_store_kwargs={"chat_history_id": "user_1"},
)
print(result2["last_message"].text)
```

--------------------------------

### Run Unit Tests with Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Execute unit tests using Hatch. It's recommended to run tests on specific modules or use filtering arguments like -k due to the large size of the full test suite.

```bash
hatch run test:unit
```

--------------------------------

### Open Shell with Test Dependencies using Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Access an interactive shell with test dependencies pre-loaded. This is useful for exploratory work within the project's test environment.

```bash
hatch -e test shell
```

--------------------------------

### Run Integration Tests with Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Execute integration tests via Hatch. Similar to unit tests, consider using filtering for efficiency.

```bash
hatch run test:integration
```

--------------------------------

### Initialize and Use Mem0MemoryStore

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Configure and utilize Mem0MemoryStore for long-term memory. Supports adding, searching, and deleting memories scoped by user, run, or agent ID.

```python
import os
from haystack.dataclasses import ChatMessage
from haystack_experimental.memory_stores.mem0 import Mem0MemoryStore

os.environ["MEM0_API_KEY"] = "your-mem0-api-key"

memory = Mem0MemoryStore()

# Store conversation turns as memories
messages = [
    ChatMessage.from_user("I prefer concise answers."),
    ChatMessage.from_assistant("Understood, I'll keep my answers brief."),
]
added = memory.add_memories(messages=messages, user_id="alice")
print(added)  # [{"memory_id": "...", "memory": "User prefers concise answers"}]

# Retrieve relevant memories for a new query
results = memory.search_memories(
    query="How should I format my answers?",
    user_id="alice",
    top_k=3,
)
for msg in results:
    print(msg.text)  # "User prefers concise answers."

# Get all memories as a single combined ChatMessage
combined = memory.search_memories_as_single_message(user_id="alice")
print(combined.text)
# - MEMORY #1: User prefers concise answers.

# Delete all memories for a user
memory.delete_all_memories(user_id="alice")
```

--------------------------------

### Initialize and Use LLMSummarizer

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Component for summarizing Document objects using a ChatGenerator. Controls summary detail and supports recursive summarization.

```python
from haystack import Document
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack_experimental.components.summarizers import LLMSummarizer

long_text = (
    "Machine learning is a branch of AI that enables systems to learn from data. "
    "Supervised learning uses labeled datasets to train models for prediction tasks. "
    "Unsupervised learning finds hidden structure in unlabeled data. "
    "Reinforcement learning trains agents through reward signals. "
    "Deep learning leverages multi-layer neural networks to capture complex patterns."
)

summarizer = LLMSummarizer(
    chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
    system_prompt="Rewrite this text in summarized form.",
    summary_detail=0,           # 0 = most concise
    minimum_chunk_size=500,
    summarize_recursively=False,
)
summarizer.warm_up()

docs = [Document(content=long_text)]
result = summarizer.run(documents=docs)
print(result["documents"][0].meta["summary"])
# Expected output: a concise one- or two-sentence summary

# Higher detail for a longer, more complete summary
result_detailed = summarizer.run(documents=docs, detail=0.8, summarize_recursively=True)
print(result_detailed["documents"][0].meta["summary"])
```

--------------------------------

### Import and Use Experimental Generator

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Import and instantiate an experimental generator component. Requires `ChatMessage` from `haystack.dataclasses`.

```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.components.generators import FoobarGenerator

c = FoobarGenerator()
c.run([ChatMessage.from_user("What's an experiment? Be brief.")])
```

--------------------------------

### HallucinationScoreConfig Parameters

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Configures parameters for the EDFL hallucination risk calculator. Supports default, evidence-based, and auto modes.

```python
from haystack_experimental.utils.hallucination_risk_calculator.dataclasses import HallucinationScoreConfig

# Default configuration (closed-book, conservative)
default_cfg = HallucinationScoreConfig()
# HallucinationScoreConfig(
#   skeleton_policy='closed_book', n_samples=7, m=6,
#   temperature=0.3, h_star=0.05, isr_threshold=1.0,
#   margin_extra_bits=0.2, B_clip=12.0, clip_mode='one-sided'
# )

# Evidence-based configuration with relaxed risk target
evidence_cfg = HallucinationScoreConfig(
    skeleton_policy="evidence_erase",
    n_samples=10,
    m=8,
    temperature=0.4,
    h_star=0.10,         # allow up to 10% hallucination risk
    isr_threshold=0.9,
    margin_extra_bits=0.1,
)

# Auto mode — automatically selects skeleton strategy
auto_cfg = HallucinationScoreConfig(skeleton_policy="auto")
```

--------------------------------

### Format and Lint Code with Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Apply code formatting and linting checks using the 'fmt' command within Hatch. This ensures code consistency and adherence to project standards.

```bash
hatch run fmt
```

--------------------------------

### Extended AgentSnapshot for HITL

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Carries `tool_execution_decisions` for breakpoint-based human-in-the-loop workflows. Supports manual construction and serialization.

```python
from haystack.dataclasses.breakpoints import AgentBreakpoint, ToolBreakpoint
from haystack_experimental.dataclasses.breakpoints import AgentSnapshot
from haystack.human_in_the_loop.dataclasses import ToolExecutionDecision
from datetime import datetime

# Manually construct (normally created internally by the Agent)
snapshot = AgentSnapshot(
    component_inputs={},
    component_visits={"chat_generator": 1, "tool_invoker": 1},
    break_point=AgentBreakpoint(
        agent_name="my_agent",
        break_point=ToolBreakpoint(
            component_name="tool_invoker",
            tool_name="delete_file",
            visit_count=1,
            snapshot_file_path="./snapshots",
        ),
    ),
    timestamp=datetime.now(),
    tool_execution_decisions=None,
)

# Serialize / deserialize
d = snapshot.to_dict()
restored = AgentSnapshot.from_dict(d)

# Attach human decision before resuming
decision = ToolExecutionDecision(tool_call_id="call_abc", decision="approve")
restored.tool_execution_decisions = [decision]
```

--------------------------------

### Agent.run_async for Asynchronous Execution

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Illustrates the asynchronous version of the Agent's run method, suitable for non-blocking operations and async streaming. It accepts the same parameters as `run` and returns the same output structure.

```python
import asyncio
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import create_tool_from_function
from haystack_experimental.components.agents import Agent

def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o"),
    tools=[create_tool_from_function(add)],
)
agent.warm_up()

async def main():
    result = await agent.run_async(
        messages=[ChatMessage.from_user("What is 17.5 + 4.2?")],
    )
    print(result["last_message"].text)

asyncio.run(main())
```

--------------------------------

### Import Experimental Pipeline

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Override the default Haystack Pipeline with an experimental version by changing the import statement. This allows opting into experimental pipeline features.

```python
# from haystack import Pipeline
from haystack_experimental import Pipeline

pipe = Pipeline()
# ...
pipe.run(...)
```

--------------------------------

### BreakpointConfirmationStrategy for Pausing Execution

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Shows how to use BreakpointConfirmationStrategy to pause agent execution when a tool requires human review. The agent's state is saved to a snapshot file, allowing the workflow to be resumed later with an approved or modified tool decision.

```python
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.core.errors import BreakpointException
from haystack.core.pipeline.breakpoint import load_pipeline_snapshot
from haystack.dataclasses import ChatMessage
from haystack.human_in_the_loop import AlwaysAskPolicy, BlockingConfirmationStrategy, RichConsoleUI, ToolExecutionDecision
from haystack.tools import create_tool_from_function

from haystack_experimental.components.agents import Agent
from haystack_experimental.components.agents.human_in_the_loop import BreakpointConfirmationStrategy
from haystack_experimental.components.agents.human_in_the_loop.breakpoint import (
    get_tool_calls_and_descriptions_from_snapshot,
)

def delete_file(path: str) -> str:
    """Delete a file at the given path."""
    return f"Deleted {path}"

delete_tool = create_tool_from_function(delete_file)

SNAPSHOT_DIR = "./snapshots"
agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4.1"),
    tools=[delete_tool],
    confirmation_strategies={
        delete_tool.name: BreakpointConfirmationStrategy(snapshot_file_path=SNAPSHOT_DIR)
    },
)
agent.warm_up()

# Step 1: initial run — pauses at the tool call
try:
    result = agent.run(messages=[ChatMessage.from_user("Delete /tmp/old_data.csv")])
except BreakpointException as e:
    print("Paused:", e)

# Step 2: load snapshot and present tool call to a human
import os
snapshot_file = max(
    [os.path.join(SNAPSHOT_DIR, f) for f in os.listdir(SNAPSHOT_DIR)],
    key=os.path.getctime,
)
pipeline_snapshot = load_pipeline_snapshot(snapshot_file)
tool_calls, descriptions = get_tool_calls_and_descriptions_from_snapshot(
    agent_snapshot=pipeline_snapshot.agent_snapshot, breakpoint_tool_only=True
)
```

--------------------------------

### Initialize MarkdownHeaderLevelInferrer

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Haystack preprocessor component to normalize Markdown header hierarchies in Document objects. The first header is always level 1.

```python
from haystack import Document
from haystack_experimental.components.preprocessors import MarkdownHeaderLevelInferrer
```

--------------------------------

### Extract Tool Calls from AgentSnapshot

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Utility function to extract tool calls and their descriptions from an AgentSnapshot at a ToolBreakpoint. Reconstructs arguments for human approval.

```python
from haystack.core.pipeline.breakpoint import load_pipeline_snapshot
from haystack_experimental.components.agents.human_in_the_loop.breakpoint import (
    get_tool_calls_and_descriptions_from_snapshot,
)

snapshot = load_pipeline_snapshot("./snapshots/snapshot_20250601_120000.json")

# Only the tool that caused the breakpoint
tool_calls, descriptions = get_tool_calls_and_descriptions_from_snapshot(
    agent_snapshot=snapshot.agent_snapshot,
    breakpoint_tool_only=True,
)
for tc in tool_calls:
    print(f"Tool: {tc['tool_name']}")
    print(f"Args: {tc['arguments']}")
    print(f"Description: {descriptions[tc['tool_name']]}")
```

--------------------------------

### Implement Experimental Generator Component

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Define a new experimental generator component within the `haystack_experimental.components.generators` namespace. Uses the `@component` decorator from `haystack.component`.

```python
# in haystack_experimental/components/generators/foobar.py

from haystack import component


@component
class FoobarGenerator:
    ...
```

--------------------------------

### Manage Chat Messages with InMemoryChatMessageStore

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Use InMemoryChatMessageStore for storing and retrieving chat messages. Supports configurable windowing, deduplication, and serialization. It can be cleared entirely or by session.

```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore

store = InMemoryChatMessageStore(skip_system_messages=True, last_k=10)

# Write messages
messages = [
    ChatMessage.from_assistant("Hello! How can I help you?"),
    ChatMessage.from_user("What is a Python Protocol?"),
]
written = store.write_messages(chat_history_id="session_42", messages=messages)
print(written)  # 2

# Count
print(store.count_messages("session_42"))  # 2

# Retrieve last 1 round
retrieved = store.retrieve_messages(chat_history_id="session_42", last_k=1)
print(retrieved[-1].text)  # "What is a Python Protocol?"

# Delete a single session
store.delete_messages("session_42")
print(store.count_messages("session_42"))  # 0

# Delete everything
store.delete_all_messages()

# Serialization round-trip
d = store.to_dict()
store2 = InMemoryChatMessageStore.from_dict(d)
```

--------------------------------

### Infer Markdown Header Levels

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Infers header levels for Markdown documents. Handles single and multiple documents.

```python
from haystack.dataclasses import Document
from haystack_experimental.components.rankers import MarkdownHeaderLevelInferrer

flat_md = "## Introduction
## Section A
Some content here.
## Section B
## Subsection B1"
doc = Document(content=flat_md)

inferrer = MarkdownHeaderLevelInferrer()
result = inferrer.run(documents=[doc])
print(result["documents"][0].content)
# # Introduction
# ## Section A
# Some content here.
# ## Section B
# ### Subsection B1

# Multiple documents processed in one call
docs = [Document(content="## Title
## Sub
Text
## Sub2"), Document(content="## A
## B")]
out = inferrer.run(documents=docs)
for d in out["documents"]:
    print(d.content)
    print("---")
```

--------------------------------

### Retrieve and Merge Chat History with ChatMessageRetriever

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

ChatMessageRetriever fetches chat messages from a store and merges them with current messages. System messages are prioritized, followed by history, and then new non-system messages.

```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore
from haystack_experimental.components.retrievers import ChatMessageRetriever

store = InMemoryChatMessageStore()
store.write_messages(
    chat_history_id="session_99",
    messages=[
        ChatMessage.from_user("What is the capital of France?"),
        ChatMessage.from_assistant("The capital of France is Paris."),
    ],
)

retriever = ChatMessageRetriever(chat_message_store=store, last_k=10)

# Retrieve history and merge with a new incoming user message
new_message = ChatMessage.from_user("What about Germany?")
result = retriever.run(
    chat_history_id="session_99",
    current_messages=[new_message],
)
# result["messages"] = [<stored user>, <stored assistant>, <new user>]
for msg in result["messages"]:
    print(f"[{msg.role}] {msg.text}")
```

--------------------------------

### Delete Hatch Environment

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Prune and remove unused or old Hatch environments to free up disk space. Use this command to maintain a clean development environment.

```bash
hatch env prune
```

--------------------------------

### Approve Tool Call with AgentSnapshot

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Approve a tool call made by an agent by updating the pipeline snapshot with a ToolExecutionDecision. This is used in a human-in-the-loop workflow.

```python
from haystack.core.pipeline.breakpoint import ToolExecutionDecision

approved = ToolExecutionDecision(tool_call_id=tool_calls[0]["id"], decision="approve")
pipeline_snapshot.agent_snapshot.tool_execution_decisions = [approved]
result = agent.run(
    messages=[],
    snapshot=pipeline_snapshot.agent_snapshot,
)
print(result["last_message"].text)
```

--------------------------------

### Write Chat Messages to Store using ChatMessageWriter

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

The ChatMessageWriter component writes a list of ChatMessage objects to a specified ChatMessageStore. It returns the count of messages successfully written.

```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore
from haystack_experimental.components.writers import ChatMessageWriter

store = InMemoryChatMessageStore()
writer = ChatMessageWriter(chat_message_store=store)

messages = [
    ChatMessage.from_user("Tell me about RAG."),
    ChatMessage.from_assistant("RAG stands for Retrieval-Augmented Generation..."),
]
result = writer.run(chat_history_id="user_7_session_1", messages=messages)
print(result)  # {"messages_written": 2}
```

--------------------------------

### OpenAIChatGenerator with Hallucination Scoring

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

Generates chat responses with optional hallucination risk evaluation using the EDFL framework. Requires `hallucination_score_config` for scoring.

```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.components.generators.chat import OpenAIChatGenerator
from haystack_experimental.utils.hallucination_risk_calculator.dataclasses import HallucinationScoreConfig

llm = OpenAIChatGenerator(model="gpt-4o")

# RAG / evidence-based usage
rag_prompt = (
    "Task: Answer strictly based on the evidence provided below.

```

```python
"Question: Who won the Nobel Prize in Physics in 2019?

```

```python
"Evidence:

```

```python
'- Nobel Prize press release (2019): James Peebles (1/2); Michel Mayor & Didier Queloz (1/2).

```

```python
"Constraints: If evidence is insufficient or conflicting, refuse."


```

```python
result = llm.run(
    messages=[ChatMessage.from_user(rag_prompt)],
    hallucination_score_config=HallucinationScoreConfig(
        skeleton_policy="evidence_erase",  # erases evidence fields to estimate prior
        n_samples=7,
        m=6,
        temperature=0.3,
        h_star=0.05,
    ),
)
reply = result["replies"][0]
print(f"Decision : {reply.meta['hallucination_decision']}")   # ANSWER or REFUSE
print(f"Risk bound: {reply.meta['hallucination_risk']:.3f}")  # EDFL upper bound
print(f"Rationale : {reply.meta['hallucination_rationale']}")
print(f"Answer    : {reply.text}")

# Closed-book usage (no evidence in prompt)
cb_result = llm.run(
    messages=[ChatMessage.from_user("What is the boiling point of water at sea level?")],
    hallucination_score_config=HallucinationScoreConfig(skeleton_policy="closed_book"),
)
print(cb_result["replies"][0].meta["hallucination_decision"])  # ANSWER
```

--------------------------------

### Perform Type Checking with mypy using Hatch

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/AGENTS.md

Run static type checking using mypy through Hatch. When fixing type issues, prefer code modifications over type ignores, casts, or assertions unless absolutely necessary, in which case, provide a justification.

```bash
hatch run test:types
```

--------------------------------

### InMemoryChatMessageStore

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

An in-memory store for ChatMessage objects, keyed by chat_history_id. It supports configurable windowing, automatic deduplication, and optional system-message filtering. Messages can be written, counted, retrieved, and deleted.

```APIDOC
## InMemoryChatMessageStore

### Description
In-memory store for `ChatMessage` objects, keyed by a `chat_history_id` string. Supports configurable windowing (`last_k`), automatic deduplication by message ID, and optional system-message filtering. Can be serialized and deserialized with `to_dict` / `from_dict`.

### Methods
- `write_messages(chat_history_id: str, messages: list[ChatMessage]) -> int`
- `count_messages(chat_history_id: str) -> int`
- `retrieve_messages(chat_history_id: str, last_k: int) -> list[ChatMessage]`
- `delete_messages(chat_history_id: str)`
- `delete_all_messages()`
- `to_dict() -> dict`
- `from_dict(data: dict) -> InMemoryChatMessageStore`

### Example
```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore

store = InMemoryChatMessageStore(skip_system_messages=True, last_k=10)

# Write messages
messages = [
    ChatMessage.from_assistant("Hello! How can I help you?"),
    ChatMessage.from_user("What is a Python Protocol?"),
]
written = store.write_messages(chat_history_id="session_42", messages=messages)
print(written)  # 2

# Count
print(store.count_messages("session_42"))  # 2

# Retrieve last 1 round
retrieved = store.retrieve_messages(chat_history_id="session_42", last_k=1)
print(retrieved[-1].text)  # "What is a Python Protocol?"

# Delete a single session
store.delete_messages("session_42")
print(store.count_messages("session_42"))  # 0

# Delete everything
store.delete_all_messages()

# Serialization round-trip
d = store.to_dict()
store2 = InMemoryChatMessageStore.from_dict(d)
```
```

--------------------------------

### Override Existing Pipeline with Experimental Features

Source: https://github.com/deepset-ai/haystack-experimental/blob/main/README.md

Subclass the original Haystack Pipeline to introduce experimental methods like `run_async` or modify existing ones like `to_dict` by adding new parameters.

```python
# in haystack_experiment/src/haystack_experiment/core/pipeline/pipeline.py

from haystack.core.pipeline import Pipeline as HaystackPipeline


class Pipeline(HaystackPipeline):
    # Any new experimental method that doesn't exist in the original class
    def run_async(self, inputs) -> Dict[str, Dict[str, Any]]:
        ...

    # Existing methods with breaking changes to their signature, like adding a new mandatory param
    def to_dict(self, new_param: str) -> Dict[str, Any]:
        # do something with the new parameter
        print(new_param)
        # call the original method
        return super().to_dict()
```

--------------------------------

### ChatMessageRetriever

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

A Haystack pipeline component designed to retrieve stored ChatMessage objects from a ChatMessageStore. It merges the retrieved history with current messages, prioritizing system messages, then history, and finally new non-system messages.

```APIDOC
## ChatMessageRetriever

### Description
Haystack pipeline component that retrieves stored `ChatMessage` objects from a `ChatMessageStore`. Merges retrieved history with optional current messages, placing system messages first, history second, and new non-system messages last.

### Parameters
- `chat_message_store`: The `ChatMessageStore` instance to retrieve messages from.
- `last_k`: The number of recent messages to retrieve.

### Methods
- `run(chat_history_id: str, current_messages: list[ChatMessage] = None) -> dict`

### Example
```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore
from haystack_experimental.components.retrievers import ChatMessageRetriever

store = InMemoryChatMessageStore()
store.write_messages(
    chat_history_id="session_99",
    messages=[
        ChatMessage.from_user("What is the capital of France?"),
        ChatMessage.from_assistant("The capital of France is Paris."),
    ],
)

retriever = ChatMessageRetriever(chat_message_store=store, last_k=10)

# Retrieve history and merge with a new incoming user message
new_message = ChatMessage.from_user("What about Germany?")
result = retriever.run(
    chat_history_id="session_99",
    current_messages=[new_message],
)
# result["messages"] = [<stored user>, <stored assistant>, <new user>]
for msg in result["messages"]:
    print(f"[{msg.role}] {msg.text}")
```
```

--------------------------------

### ChatMessageWriter

Source: https://context7.com/deepset-ai/haystack-experimental/llms.txt

A Haystack pipeline component that writes a list of ChatMessage objects to a ChatMessageStore under a specified chat_history_id. It returns the count of messages successfully written.

```APIDOC
## ChatMessageWriter

### Description
Haystack pipeline component that writes a list of `ChatMessage` objects to a `ChatMessageStore` under a given `chat_history_id`. Returns the number of messages written.

### Parameters
- `chat_message_store`: The `ChatMessageStore` instance to write messages to.

### Methods
- `run(chat_history_id: str, messages: list[ChatMessage]) -> dict`

### Example
```python
from haystack.dataclasses import ChatMessage
from haystack_experimental.chat_message_stores.in_memory import InMemoryChatMessageStore
from haystack_experimental.components.writers import ChatMessageWriter

store = InMemoryChatMessageStore()
writer = ChatMessageWriter(chat_message_store=store)

messages = [
    ChatMessage.from_user("Tell me about RAG."),
    ChatMessage.from_assistant("RAG stands for Retrieval-Augmented Generation..."),
]
result = writer.run(chat_history_id="user_7_session_1", messages=messages)
print(result)  # {"messages_written": 2}
```
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.