Module agents.llm.llm

Classes

class ConversationalGraphResponse (**data: Any)

Expand source code

class ConversationalGraphResponse(BaseModel):
    """ Data model to hold Conversational Graph response data."""
    
    response_to_user:str = Field(..., description="Response to the user by agent")
    extracted_values:List[ExtractedField] = Field(default_factory=list, description="List of extracted values from the user input")
    move_forward:bool = Field(False, description="If we want to Move forward to the next state")
    reasoning:str = Field("", description="Reasoning for the response")
    chosen_branch:str = Field(None, description="Chosen branch for the move forward")
    is_off_topic:bool = Field(False, description="Is the user input off topic")
    backtrack_to_state:str = Field(None, description="Backtrack to the state")
    current_state_id:str = Field(None, description="exact state_id of current state")

Data model to hold Conversational Graph response data.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var backtrack_to_state : str

var chosen_branch : str

var current_state_id : str

var extracted_values : List[ExtractedField]

var is_off_topic : bool

var model_config

var move_forward : bool

var reasoning : str

var response_to_user : str

class ExtractedField (**data: Any)

Expand source code

class ExtractedField(BaseModel):
    key: str = Field(..., description="The name of the field")
    value: Union[str, int, float, bool] = Field(..., description="The value of the field")

Usage Documentation

Models

A base class for creating Pydantic models.

Attributes

__class_vars__

The names of the class variables defined on the model.

__private_attributes__

Metadata about the private attributes of the model.

__signature__

The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__

Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__

The core schema of the model.

__pydantic_custom_init__

Whether the model has a custom __init__ function.

__pydantic_decorators__

Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__

Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.

__pydantic_parent_namespace__

Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__

The name of the post-init method for the model, if defined.

__pydantic_root_model__

Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__

The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__

The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__

A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__

A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__

A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.

__pydantic_fields_set__

The names of fields explicitly set during instantiation.

__pydantic_private__

Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var key : str

var model_config

var value : str | int | float | bool

class LLM

Expand source code

class LLM(EventEmitter[Literal["error"]]):
    """
    Base class for LLM implementations.
    """

    def __init__(self) -> None:
        """
        Initialize the LLM base class.
        """
        super().__init__()
        self._label = f"{type(self).__module__}.{type(self).__name__}"

    @property
    def label(self) -> str:
        """
        Get the LLM provider label.

        Returns:
            str: A string identifier for the LLM provider (e.g., "videosdk.plugins.openai.llm.OpenAILLM").
        """
        return self._label

    @abstractmethod
    async def chat(
        self,
        messages: ChatContext,
        tools: list[FunctionTool] | None = None,
        **kwargs: Any
    ) -> AsyncIterator[LLMResponse]:
        """
        Main method to interact with the LLM.

        Args:
            messages (ChatContext): The conversation context containing message history.
            tools (list[FunctionTool] | None, optional): List of available function tools for the LLM to use.
            **kwargs (Any): Additional arguments specific to the LLM provider implementation.

        Returns:
            AsyncIterator[LLMResponse]: An async iterator yielding LLMResponse objects as they're generated.

        Raises:
            NotImplementedError: This method must be implemented by subclasses.
        """
        raise NotImplementedError

    @abstractmethod
    async def cancel_current_generation(self) -> None:
        """
        Cancel the current LLM generation if active.

        Raises:
            NotImplementedError: This method must be implemented by subclasses.
        """
        # override in subclasses
        pass

    async def aclose(self) -> None:
        """
        Cleanup resources.
        """
        logger.info(f"Cleaning up LLM: {self.label}")
        
        await self.cancel_current_generation()
        
        try:
            import gc
            gc.collect()
            logger.info(f"LLM garbage collection completed: {self.label}")
        except Exception as e:
            logger.error(f"Error during LLM garbage collection: {e}")
        
        logger.info(f"LLM cleanup completed: {self.label}")

    async def __aenter__(self) -> LLM:
        """
        Async context manager entry point.
        """
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """
        Async context manager exit point.
        """
        await self.aclose()

Base class for LLM implementations.

Initialize the LLM base class.

Ancestors

• EventEmitter
• typing.Generic

Subclasses

• FallbackLLM

Instance variables

prop label : str

Expand source code

@property
def label(self) -> str:
    """
    Get the LLM provider label.

    Returns:
        str: A string identifier for the LLM provider (e.g., "videosdk.plugins.openai.llm.OpenAILLM").
    """
    return self._label

Get the LLM provider label.

Returns

str

A string identifier for the LLM provider (e.g., "videosdk.plugins.openai.llm.OpenAILLM").

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """
    Cleanup resources.
    """
    logger.info(f"Cleaning up LLM: {self.label}")
    
    await self.cancel_current_generation()
    
    try:
        import gc
        gc.collect()
        logger.info(f"LLM garbage collection completed: {self.label}")
    except Exception as e:
        logger.error(f"Error during LLM garbage collection: {e}")
    
    logger.info(f"LLM cleanup completed: {self.label}")

Cleanup resources.

async def cancel_current_generation(self) ‑> None

Expand source code

@abstractmethod
async def cancel_current_generation(self) -> None:
    """
    Cancel the current LLM generation if active.

    Raises:
        NotImplementedError: This method must be implemented by subclasses.
    """
    # override in subclasses
    pass

Cancel the current LLM generation if active.

Raises

NotImplementedError

This method must be implemented by subclasses.

async def chat(self,
messages: ChatContext,
tools: list[FunctionTool] | None = None,
**kwargs: Any) ‑> AsyncIterator[LLMResponse]

Expand source code

@abstractmethod
async def chat(
    self,
    messages: ChatContext,
    tools: list[FunctionTool] | None = None,
    **kwargs: Any
) -> AsyncIterator[LLMResponse]:
    """
    Main method to interact with the LLM.

    Args:
        messages (ChatContext): The conversation context containing message history.
        tools (list[FunctionTool] | None, optional): List of available function tools for the LLM to use.
        **kwargs (Any): Additional arguments specific to the LLM provider implementation.

    Returns:
        AsyncIterator[LLMResponse]: An async iterator yielding LLMResponse objects as they're generated.

    Raises:
        NotImplementedError: This method must be implemented by subclasses.
    """
    raise NotImplementedError

Main method to interact with the LLM.

Args

messages : ChatContext

The conversation context containing message history.

tools : list[FunctionTool] | None, optional

List of available function tools for the LLM to use.

**kwargs : Any

Additional arguments specific to the LLM provider implementation.

Returns

AsyncIterator[LLMResponse]

An async iterator yielding LLMResponse objects as they're generated.

Raises

NotImplementedError

This method must be implemented by subclasses.

Inherited members

• EventEmitter:
  • emit
  • off
  • on

class LLMResponse (**data: Any)

Expand source code

class LLMResponse(BaseModel):
    """
    Data model to hold LLM response data.

    Attributes:
        content (str): The text content generated by the LLM.
        role (ChatRole): The role of the response (typically ASSISTANT).
        metadata (Optional[dict[str, Any]]): Additional response metadata from the LLM provider.
    """
    content: str
    role: ChatRole
    metadata: Optional[dict[str, Any]] = None

Data model to hold LLM response data.

Attributes

content : str

The text content generated by the LLM.

role : ChatRole

The role of the response (typically ASSISTANT).

metadata : Optional[dict[str, Any]]

Additional response metadata from the LLM provider.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var content : str

var metadata : dict[str, typing.Any] | None

var model_config

var role : ChatRole

class ResponseChunk (content: str, metadata: dict[str, Any] | None = None, role: str | None = None)

Expand source code

class ResponseChunk(str):
    """A string subclass representing a single chunk of an LLM response, carrying optional metadata and role."""
    def __new__(cls, content: str, metadata: dict[str, Any] | None = None, role: str | None = None):
        obj = super().__new__(cls, content or "")
        obj.metadata = metadata
        obj.role = role
        return obj

    @property
    def content(self) -> str:
        return str(self)

A string subclass representing a single chunk of an LLM response, carrying optional metadata and role.

Ancestors

• builtins.str

Instance variables

prop content : str

Expand source code

@property
def content(self) -> str:
    return str(self)