sentient-agent-framework 0.3.0

This python package provides an agent framework that can be used to build agents that serve Sentient Chat events.

Sentient Agent Framework

[!WARNING] This python package is currently in beta and will likely change. It is not yet ready for production use.

In addition to supporting OpenAI API compatible agents, Sentient Chat supports a custom, open source event system for agent responses. These events can be rendered in Sentient Chat to provide a richer user experience. This particularly useful for streaming responses from an AI agent, when you might want to show the agent's work while the response is being generated, rather than having the user wait for the final response. This python package provides an agent framework that can be used to build agents that serve Sentient Chat events.

Examples of agents that use this framework/package can be found here. A client for testing agents built with this framework is available here.

Installation

pip install sentient-agent-framework

Usage

The simplest way to use this framework is to import and use the AbstractAgent class and the DefaultServer class.

AbstractAgent

The AbstractAgent class is lightweight and extensible. To use it, simply subclass the class and implement the assist() method. Use the ResponseHandler object passed to the assist() method to emit events to the client.

DefaultServer

The DefaultServer class is designed to be used with the AbstractAgent class. A concrete implementation of the AbstractAgent class is passed into the DefaultServer constructor. The DefaultServer provides SSE server with /assist endpoint and automatically streams events emitted in the assist() method to the client.

[!NOTE]
The snippet below comes from the Sentient Agent Framework Examples repository.

Example

import logging
import os
from dotenv import load_dotenv
from src.search_agent.providers.model_provider import ModelProvider
from src.search_agent.providers.search_provider import SearchProvider
from sentient_agent_framework import (
    AbstractAgent,
    DefaultServer,
    Session,
    Query,
    ResponseHandler)
from typing import AsyncIterator


load_dotenv()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class SearchAgent(AbstractAgent):
    def __init__(
            self,
            name: str
    ):
        super().__init__(name)

        model_api_key = os.getenv("MODEL_API_KEY")
        if not model_api_key:
            raise ValueError("MODEL_API_KEY is not set")
        self._model_provider = ModelProvider(api_key=model_api_key)

        search_api_key = os.getenv("TAVILY_API_KEY")
        if not search_api_key:
            raise ValueError("TAVILY_API_KEY is not set") 
        self._search_provider = SearchProvider(api_key=search_api_key)


    # Implement the assist method as required by the AbstractAgent class
    async def assist(
            self,
            session: Session,
            query: Query,
            response_handler: ResponseHandler
    ):
        """Search the internet for information."""
        # Search for information
        await response_handler.emit_text_block(
            "SEARCH", "Searching internet for results..."
        )
        search_results = await self._search_provider.search(query.prompt)
        if len(search_results["results"]) > 0:
            # Use response handler to emit JSON to the client
            await response_handler.emit_json(
                "SOURCES", {"results": search_results["results"]}
            )
        if len(search_results["images"]) > 0:
            # Use response handler to emit JSON to the client
            await response_handler.emit_json(
                "IMAGES", {"images": search_results["images"]}
            )

        # Process search results
        # Use response handler to create a text stream to stream the final 
        # response to the client
        final_response_stream = response_handler.create_text_stream(
            "FINAL_RESPONSE"
            )
        async for chunk in self.__process_search_results(search_results["results"]):
            # Use the text stream to emit chunks of the final response to the client
            await final_response_stream.emit_chunk(chunk)
        # Mark the text stream as complete
        await final_response_stream.complete()
        # Mark the response as complete
        await response_handler.complete()
    

    async def __process_search_results(
            self, 
            search_results: dict
    ) -> AsyncIterator[str]:
        """Process the search results."""
        process_search_results_query = f"Summarise the following search results: {search_results}"
        async for chunk in self._model_provider.query_stream(process_search_results_query):
            yield chunk


if __name__ == "__main__":
    # Create an instance of a SearchAgent
    agent = SearchAgent(name="Search Agent")
    # Create a server to handle requests to the agent
    server = DefaultServer(agent)
    # Run the server
    server.run()

Emitting events

Whether using the AbstractAgent or the DefaultResponseHandler, a ResponseHandler is created for every agent query and is used to emit events to the client.

Emitting text events

Text events are used to send single, complete messages to the client:

await response_handler.emit_text_block(
    "PLAN", "Rephrasing user query..."
)

Emitting JSON events

JSON events are used to send JSON objects to the client:

await response_handler.emit_json(
    "SOURCES", {"results": search_results["results"]}
)

Emitting error events

Error events are used to send error messages to the client:

await response_handler.emit_error(
    "ERROR", {"message": "An error occurred"}
)

Completing a response

At the end of a response, response_handler.complete() is called to signal the end of the response (this will emit a DoneEvent using the Hook):

await response_handler.complete()

Emitting a stream of text chunks

To stream a longer response one chunk at a time, use the response_handler.create_text_stream method. This returns a StreamEventEmitter that can be used to stream text to the client using the emit_chunk method:

final_response_stream = response_handler.create_text_stream(
    "FINAL_RESPONSE"
    )
for chunk in self.__process_search_results(search_results["results"]):
    await final_response_stream.emit_chunk(chunk)

Completing a stream

At the end of the stream, final_response_stream.complete() is called to signal the end of the stream (this will emit a TextChunkEvent with is_complete=True):

await final_response_stream.complete()

Documentation

• Interface Documentation
• Implementation Documentation