Chat Interface

The ChatInterface is the main interface for the chat service. It defines the core functionality required for a chat service that can return various types of responses such as:

• Text: Regular text responses streamed chunk by chunk
• References: Source documents used to generate the answer

ragbits.chat.interface.ChatInterface

Bases: ABC

Base interface for chat implementations.

This interface defines the core functionality required for a chat service that can return various types of responses such as:

• Text: Regular text responses streamed chunk by chunk
• References: Source documents used to generate the answer
• State updates: Updates to the conversation state

feedback_config class-attribute instance-attribute

feedback_config: FeedbackConfig = FeedbackConfig()

history_persistence class-attribute instance-attribute

history_persistence: HistoryPersistenceStrategy | None = None

create_text_response staticmethod

create_text_response(text: str) -> ChatResponse

Helper method to create a text response.

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

@staticmethod
def create_text_response(text: str) -> ChatResponse:
    """Helper method to create a text response."""
    return ChatResponse(type=ChatResponseType.TEXT, content=text)

create_reference staticmethod

create_reference(title: str, content: str, url: str | None = None) -> ChatResponse

Helper method to create a reference response.

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

@staticmethod
def create_reference(
    title: str,
    content: str,
    url: str | None = None,
) -> ChatResponse:
    """Helper method to create a reference response."""
    return ChatResponse(
        type=ChatResponseType.REFERENCE,
        content=Reference(title=title, content=content, url=url),
    )

create_state_update staticmethod

create_state_update(state: dict[str, Any]) -> ChatResponse

Helper method to create a state update response with signature.

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

@staticmethod
def create_state_update(state: dict[str, Any]) -> ChatResponse:
    """Helper method to create a state update response with signature."""
    signature = ChatInterface._sign_state(state)
    return ChatResponse(
        type=ChatResponseType.STATE_UPDATE,
        content=StateUpdate(state=state, signature=signature),
    )

verify_state staticmethod

verify_state(state: dict[str, Any], signature: str) -> bool

Verify that a state and signature match.

PARAMETER	DESCRIPTION
state	The state dictionary to verify TYPE: dict[str, Any]
signature	The signature to check against TYPE: str

RETURNS	DESCRIPTION
bool	True if the signature is valid, False otherwise

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

@staticmethod
def verify_state(state: dict[str, Any], signature: str) -> bool:
    """
    Verify that a state and signature match.

    Args:
        state: The state dictionary to verify
        signature: The signature to check against

    Returns:
        True if the signature is valid, False otherwise
    """
    expected_signature = ChatInterface._sign_state(state)
    return hmac.compare_digest(expected_signature, signature)

setup async

setup() -> None

Setup the chat interface.

This method is called after the chat interface is initialized and before the chat method is called. It is used to setup the chat interface, such as loading the model or initializing the vector store.

This method is optional and can be overridden by subclasses.

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

async def setup(self) -> None:  # noqa: B027
    """
    Setup the chat interface.

    This method is called after the chat interface is initialized and before the chat method is called.
    It is used to setup the chat interface, such as loading the model or initializing the vector store.

    This method is optional and can be overridden by subclasses.
    """
    pass

chat abstractmethod async

chat(message: str, history: ChatFormat | None = None, context: ChatContext | None = None) -> AsyncGenerator[ChatResponse, None]

Process a chat message and yield responses asynchronously.

PARAMETER	DESCRIPTION
message	The current user message TYPE: str
history	Optional list of previous messages in the conversation TYPE: ChatFormat | None DEFAULT: None
context	Optional context containing conversation metadata and state. Will be automatically populated with message_id and conversation_id. TYPE: ChatContext | None DEFAULT: None

YIELDS	DESCRIPTION
AsyncGenerator[ChatResponse, None]	ChatResponse objects containing different types of content:
AsyncGenerator[ChatResponse, None]	Text chunks for the actual response
AsyncGenerator[ChatResponse, None]	Reference documents used to generate the response
AsyncGenerator[ChatResponse, None]	State updates when the conversation state changes

Example

chat = MyChatImplementation()
async for response in chat.chat("What is Python?"):
    if text := response.as_text():
        print(f"Text: {text}")
    elif ref := response.as_reference():
        print(f"Reference: {ref.title}")
    elif state := response.as_state_update():
        if verify_state(state.state, state.signature):
            # Update client state
            pass

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

@abstractmethod
async def chat(
    self,
    message: str,
    history: ChatFormat | None = None,
    context: ChatContext | None = None,
) -> AsyncGenerator[ChatResponse, None]:
    """
    Process a chat message and yield responses asynchronously.

    Args:
        message: The current user message
        history: Optional list of previous messages in the conversation
        context: Optional context containing conversation metadata and state.
                Will be automatically populated with message_id and conversation_id.

    Yields:
        ChatResponse objects containing different types of content:
        - Text chunks for the actual response
        - Reference documents used to generate the response
        - State updates when the conversation state changes

    Example:
        ```python
        chat = MyChatImplementation()
        async for response in chat.chat("What is Python?"):
            if text := response.as_text():
                print(f"Text: {text}")
            elif ref := response.as_reference():
                print(f"Reference: {ref.title}")
            elif state := response.as_state_update():
                if verify_state(state.state, state.signature):
                    # Update client state
                    pass
        ```
    """
    yield ChatResponse(type=ChatResponseType.TEXT, content="Ragbits cannot respond - please implement chat method!")
    raise NotImplementedError("Chat implementations must implement chat method")

save_feedback async

save_feedback(message_id: str, feedback: Literal['like', 'dislike'], payload: dict) -> None

Save feedback about a chat message.

PARAMETER	DESCRIPTION
message_id	The ID of the message TYPE: str
feedback	The type of feedback TYPE: Literal['like', 'dislike']
payload	The payload of the feedback TYPE: dict

Source code in packages/ragbits-chat/src/ragbits/chat/interface/_interface.py

async def save_feedback(
    self,
    message_id: str,
    feedback: Literal["like", "dislike"],
    payload: dict,
) -> None:
    """
    Save feedback about a chat message.

    Args:
        message_id: The ID of the message
        feedback: The type of feedback
        payload: The payload of the feedback
    """
    logger.info(f"[{self.__class__.__name__}] Saving {feedback} for message {message_id} with payload {payload}")