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)

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