Module haystack_experimental.components.writers.chat_message_writer

ChatMessageWriter

Writes chat messages to an underlying ChatMessageStore.

Usage example:

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

messages = [
    ChatMessage.from_assistant("Hello, how can I help you?"),
    ChatMessage.from_user("I have a question about Python."),
]
message_store = InMemoryChatMessageStore()
writer = ChatMessageWriter(message_store)
writer.run(messages)

ChatMessageWriter.__init__

def __init__(message_store: ChatMessageStore)

Create a ChatMessageWriter component.

Arguments:

• message_store: The ChatMessageStore where the chat messages are to be written.

ChatMessageWriter.to_dict

def to_dict() -> Dict[str, Any]

Serializes the component to a dictionary.

Returns:

Dictionary with serialized data.

ChatMessageWriter.from_dict

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ChatMessageWriter"

Deserializes the component from a dictionary.

Arguments:

• data: The dictionary to deserialize from.

Raises:

• DeserializationError: If the message store is not properly specified in the serialization data or its type cannot be imported.

Returns:

The deserialized component.

ChatMessageWriter.run

@component.output_types(messages_written=int)
def run(messages: List[ChatMessage])

Run the ChatMessageWriter on the given input data.

Arguments:

• messages: A list of chat messages to write to the store.

Raises:

• ValueError: If the specified message store is not found.

Returns:

• messages_written: Number of messages written to the ChatMessageStore.

Module haystack_experimental.components.writers.document_writer

DocumentWriter

Writes documents to a DocumentStore.

Usage example

from haystack import Document
from haystack_experimental.components.writers import DocumentWriter
from haystack_experimental.document_stores.in_memory import InMemoryDocumentStore

docs = [
    Document(content="Python is a popular programming language"),
]
doc_store = InMemoryDocumentStore()
writer = DocumentWriter(document_store=doc_store)
writer.run(docs)

DocumentWriter.__init__

def __init__(document_store: DocumentStore,
             policy: DuplicatePolicy = DuplicatePolicy.NONE)

Create a DocumentWriter component.

Arguments:

• document_store: The instance of the document store where you want to store your documents.
• policy: The policy to apply when a Document with the same ID already exists in the DocumentStore.
• DuplicatePolicy.NONE: Default policy, relies on the DocumentStore settings.
• DuplicatePolicy.SKIP: Skips documents with the same ID and doesn't write them to the DocumentStore.
• DuplicatePolicy.OVERWRITE: Overwrites documents with the same ID.
• DuplicatePolicy.FAIL: Raises an error if a Document with the same ID is already in the DocumentStore.

DocumentWriter.run_async

@component.output_types(documents_written=int)
async def run_async(documents: List[Document],
                    policy: Optional[DuplicatePolicy] = None)

Run the DocumentWriter on the given input data.

Arguments:

• documents: A list of documents to write to the document store.
• policy: The policy to use when encountering duplicate documents.

Raises:

• ValueError: If the specified document store is not found.

Returns:

Number of documents written to the document store.