Schema

Modules concerning the schemas used throughout the Gen AI applications.

`Chunk`

Bases: BaseModel

Represents a chunk of content retrieved from a vector store.

Attributes:

Name	Type	Description
`id`	`str`	A unique identifier for the chunk. Defaults to a random UUID.
`content`	`str \| bytes`	The content of the chunk, either text or binary.
`metadata`	`dict[str, Any]`	Additional metadata associated with the chunk. Defaults to an empty dictionary.
`score`	`float \| None`	Similarity score of the chunk (if available). Defaults to None.

`repr()`

Return a string representation of the Chunk.

Returns:

Name	Type	Description
`str`	`str`	The string representation of the Chunk.

`is_binary()`

Check if the content is binary.

Returns:

Name	Type	Description
`bool`	`bool`	True if the content is binary, False otherwise.

`is_text()`

Check if the content is text.

Returns:

Name	Type	Description
`bool`	`bool`	True if the content is text, False otherwise.

`validate_content(value)` `classmethod`

Validate the content of the Chunk.

This is a class method required by Pydantic validators. As such, it follows its signature and conventions.

Parameters:

Name	Type	Description	Default
`value`	`str \| bytes`	The content to validate.	required

Returns:

Type	Description
`str \| bytes`	str \| bytes: The validated content.

Raises:

Type	Description
`ValueError`	If the content is empty or not a string or bytes.

`Component`

An abstract base class for all components used throughout the Gen AI applications.

Every instance of Component has access to class-level _default_log_level and _logger, as detailed below. For components that require high observability, it is recommended to set _default_log_level to logging.INFO or higher.

Defining Custom Components

There are two ways to define the main execution logic for a component:

Using the @main decorator (Recommended): Decorate an async method with @main to mark it as the primary entrypoint. This is the preferred approach as it provides explicit control over the main method.

```python class MyComponent(Component): _default_log_level = logging.INFO

   @main
   async def execute(self, **kwargs: Any) -> Any:
       return "Hello from @main!"

```

Implementing _run method (Deprecated): Override the abstract _run method. This is the traditional approach and still supported.

```python class MyComponent(Component): _default_log_level = logging.INFO

   async def _run(self, **kwargs: Any) -> Any:
       return "Hello, World!"

```

The run() method resolves the main entrypoint using the following precedence: 1. Method decorated with @main in the current class. 2. Method decorated with @main in the nearest ancestor class. 3. Method named in main_method property. 4. The _run method (with deprecation warning).

Attributes:

Name	Type	Description
`run_profile`	`RunProfile`	The profile of the `_run` method. This property is used by `Pipeline` to analyze the input requirements of the component. In most cases, unless you are working with `Pipeline` and `PipelineStep`s, you will not need to use this property. Do not override this property in your subclass. You also do not need to write this attribute in your component's docstring.

`input_params: type[BaseModel] | None` `property`

Return the Pydantic model describing this component's main method input parameters.

Returns:

Type	Description
`type[BaseModel] \| None`	type[BaseModel] \| None: The cached model that mirrors the signature of the resolved main method, or `None` if no main method can be determined.

Examples:

from pydantic import ValidationError

component = SomeComponent()
ParamsModel = component.input_params
assert ParamsModel.__name__ == "SomeComponentParams"
fields = list(ParamsModel.model_fields)

# Validation with valid params
params = ParamsModel(text="hello")

# Validation catches missing required fields
try:
    invalid_params = ParamsModel()  # Missing required 'text' field
except ValidationError as e:
    print(f"Validation failed: {e.error_count()} errors")

# Argument construction
payload = params.model_dump()
result = await component.run(**payload)

`run_profile: RunProfile` `property`

Analyzes the _run method and retrieves its profile.

This property method analyzes the _run method of the class to generate a RunProfile object. It also updates the method signatures for methods that fully utilize the arguments.

Returns:

Name	Type	Description
`RunProfile`	`RunProfile`	The profile of the `_run` method, including method signatures for full-pass argument usages.

`__init_subclass__(**kwargs)`

Hook called when a subclass is created.

This validates the main_method property and checks for multiple @main decorators within the current class definition. Uses MainMethodResolver for consistent validation logic.

Note: Multiple inheritance conflicts are intentionally deferred to runtime (get_main()) to allow class definition to succeed.

Raises:

Type	Description
`AttributeError`	If main_method refers to a non-existent method.
`TypeError`	If multiple methods are decorated with @main in the same class.

`as_tool(name=None, description=None, title=None)`

Convert the component's main method into a Tool instance.

Example

from gllm_core.schema import Component, main

class MyComponent(Component):
    @main
    async def my_method(self, param: str) -> str:
        return param

component = MyComponent()
tool = component.as_tool()

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Identifier for the resulting tool. Defaults to the component class name.	`None`
`description`	`str \| None`	Summary of the tool's behavior. Defaults to None, in which case the main method's docstring is used.	`None`
`title`	`str \| None`	Optional display title for the tool. Defaults to None, in which case the component's class name is used.	`None`

Returns:

Name	Type	Description
`Tool`	`Tool`	The tool wrapping the component's main method.

Raises:

Type	Description
`RuntimeError`	If the component does not declare a main method using @main or main_method.

`get_main()` `cached` `classmethod`

Return the resolved main coroutine for this Component class.

This method resolves the main method for the Component class following the precedence rules: 1. Most derived coroutine decorated with @main. 2. Method named by __main_method__. 3. _run coroutine as a deprecated fallback.

Results are cached for performance.

Returns:

Type	Description
`Callable \| None`	Callable \| None: The coroutine that will be executed by `run()` or `None` when no entrypoint can be determined.

Raises:

Type	Description
`TypeError`	If conflicting main methods are inherited from multiple ancestors.

`run(**kwargs)` `async`

Runs the operations defined for the component.

This method emits the provided input arguments using an EventEmitter instance if available, executes the resolved main method, and emits the resulting output if the EventEmitter is provided.

The main method is resolved using the following precedence: 1. Method decorated with @main in the current class. 2. Method decorated with @main in the nearest ancestor class. 3. Method named in main_method property. 4. The _run method (with deprecation warning).

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	A dictionary of arguments to be processed. May include an `event_emitter` key with an EventEmitter instance.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	The result of the resolved main method.

Raises:

Type	Description
`TypeError`	If conflicting main methods are inherited from multiple ancestors.
`AttributeError`	If main_method refers to a non-existent method.

`Event`

Bases: BaseModel

A data class to store an event attributes.

Attributes:

Name	Type	Description
`id`	`str`	The ID of the event. Defaults to None.
`value`	`str \| dict[str, Any]`	The value of the event. Defaults to an empty string.
`level`	`EventLevel`	The severity level of the event. Defaults to EventLevel.INFO.
`type`	`str`	The type of the event. Defaults to EventType.RESPONSE.
`timestamp`	`datetime`	The timestamp of the event. Defaults to the current timestamp.
`metadata`	`dict[str, Any]`	The metadata of the event. Defaults to an empty dictionary.

`serialize_level(level)`

Serializes an EventLevel object into its string representation.

This method serializes the given EventLevel object by returning its name as a string.

Parameters:

Name	Type	Description	Default
`level`	`EventLevel`	The EventLevel object to be serialized.	required

Returns:

Name	Type	Description
`str`	`str`	The name of the EventLevel object.

`Tool`

Bases: BaseModel

Model Context Protocol (MCP)-style Tool definition.

This class represents a tool that can be used by a language model to interact with the outside world, following the Model Context Protocol (MCP) specification. Tools are defined by their name, description, input and output schemas, and an optional function implementation.

The Tool class supports flexible schema handling, accepting either: 1. Dictionary-based JSON Schema objects 2. Pydantic BaseModel classes

When a Pydantic model is provided, it is automatically converted to a JSON Schema using Pydantic's model_json_schema() method.

Supported use cases include: 1. Creating a tool with dictionary schemas for input/output 2. Creating a tool with Pydantic models for input/output 3. Using the @tool decorator to create a tool from a function's type hints

Attributes:

Name	Type	Description
`name`	`str`	A string identifier for the tool, used for programmatic access.
`description`	`str`	A human-readable description of what the tool does.
`input_schema`	`dict[str, Any] \| type[BaseModel]`	JSON Schema object or Pydantic model defining the expected parameters.
`title`	`str \| None`	Optional display name for the tool.
`output_schema`	`dict[str, Any] \| type[BaseModel] \| None`	Optional JSON Schema object or Pydantic model defining the structure of the output.
`annotations`	`dict[str, Any] \| None`	Optional additional tool information for enriching the tool definition. According to MCP, display name precedence is: title, annotations.title, then name.
`meta`	`dict[str, Any] \| None`	Optional additional metadata for internal use by the system. Unlike annotations which provide additional information about the tool for clients, meta is meant for private system-level metadata that shouldn't be exposed to end users.
`func`	`Callable`	The callable function that implements this tool's behavior.
`is_async`	`bool`	Whether the tool's function is asynchronous.

`signature: inspect.Signature` `property`

Expose the underlying function's signature for introspection.

Returns:

Type	Description
`Signature`	inspect.Signature: Signature of the underlying function, or an empty signature if missing.

`call(*args, **kwargs)`

Call the underlying function.

Mirrors the original function's call semantics: 1. If the underlying function is synchronous, returns the result directly. 2. If asynchronous, returns a coroutine that must be awaited.

Parameters:

Name	Type	Description	Default
`*args`	`Any`	Positional arguments for the function.	`()`
`**kwargs`	`Any`	Keyword arguments for the function.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	Result or coroutine depending on the underlying function.

Raises:

Type	Description
`ValueError`	If no implementation function is defined.

`from_google_adk(function_declaration, func=None)` `classmethod`

Create a Tool from a Google ADK function declaration.

Parameters:

Name	Type	Description	Default
`function_declaration`	`Any`	Google ADK function declaration to convert.	required
`func`	`Callable \| None`	Optional implementation callable for the tool.	`None`

Returns:

Name	Type	Description
`Tool`	`'Tool'`	Tool instance derived from the Google ADK definition.

`from_langchain(langchain_tool)` `classmethod`

Create a Tool from a LangChain tool instance.

Parameters:

Name	Type	Description	Default
`langchain_tool`	`Any`	LangChain tool implementation to convert.	required

Returns:

Name	Type	Description
`Tool`	`'Tool'`	Tool instance derived from the LangChain representation.

`invoke(**kwargs)` `async`

Executes the defined tool with the given parameters.

This method handles both synchronous and asynchronous underlying functions.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	The parameters to pass to the tool function. These should match the input_schema definition.	`{}`

Returns:

Name	Type	Description
`Any`	`Any`	The result of the tool execution.

Raises:

Type	Description
`ValueError`	If the tool function has not been defined.
`TypeError`	If the provided parameters don't match the expected schema.

`validate_input_schema(v)` `classmethod`

Validate and convert input_schema to JSON Schema dict if it's a Pydantic model.

Parameters:

Name	Type	Description	Default
`v`	`Any`	The input schema value (dict or Pydantic model).	required

Returns:

Name	Type	Description
`dict`		A JSON Schema dict.

Raises:

Type	Description
`ValueError`	If the input schema is not a dict or Pydantic model.

`validate_output_schema(v)` `classmethod`

Validate and convert output_schema to JSON Schema dict if it's a Pydantic model.

Parameters:

Name	Type	Description	Default
`v`	`Any`	The output schema value (dict, Pydantic model, or None).	required

Returns:

Type	Description
	dict \| None: A JSON Schema dict or None.

Raises:

Type	Description
`ValueError`	If the output schema is not None, dict, or Pydantic model.

`main(method)`

Decorate a Component method as the async main entrypoint.

Usage

Declare the coroutine that should act as the primary execution path for a Component subclass. The decorated coroutine will be resolved by Component.run() unless another subclass overrides the decoration.

Parameters:

Name	Type	Description	Default
`method`	`Callable`	Coroutine to mark as the main entrypoint.	required

Returns:

Name	Type	Description
`Callable`	`Callable`	The same coroutine that is passed to the decorator. The decorator only marks the method as the main entrypoint. It does not wrap or change its behavior or signature.

Raises:

Type	Description
`TypeError`	If the decorated callable is not asynchronous.

Schema

Chunk

__repr__()

is_binary()

is_text()

validate_content(value) classmethod

Component

input_params: type[BaseModel] | None property

run_profile: RunProfile property

__init_subclass__(**kwargs)

as_tool(name=None, description=None, title=None)

get_main() cached classmethod

run(**kwargs) async

Event

serialize_level(level)

Tool

__signature__: inspect.Signature property

__call__(*args, **kwargs)

from_google_adk(function_declaration, func=None) classmethod

from_langchain(langchain_tool) classmethod

invoke(**kwargs) async

validate_input_schema(v) classmethod

validate_output_schema(v) classmethod

main(method)

`Chunk`

`repr()`

`is_binary()`

`is_text()`

`validate_content(value)` `classmethod`

`Component`

`input_params: type[BaseModel] | None` `property`

`run_profile: RunProfile` `property`

`__init_subclass__(**kwargs)`

`as_tool(name=None, description=None, title=None)`

`get_main()` `cached` `classmethod`

`run(**kwargs)` `async`

`Event`

`serialize_level(level)`

`Tool`

`signature: inspect.Signature` `property`

`call(*args, **kwargs)`

`from_google_adk(function_declaration, func=None)` `classmethod`

`from_langchain(langchain_tool)` `classmethod`

`invoke(**kwargs)` `async`

`validate_input_schema(v)` `classmethod`

`validate_output_schema(v)` `classmethod`

`main(method)`