Skip to content

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}")