types

Functional data classes to support functional components like Generator, Retriever, and Assistant.

Functions

get_model_args(model_type)

Get the required keys in model_kwargs for a specific model type.

Classes

AssistantResponse(response_str[, metadata])
CompletionUsage([completion_tokens, ...])	In sync with OpenAI completion usage api spec at openai/types/completion_usage.py
Conversation(id, name, user_id, ...)	A conversation manages the dialog turns in a whole conversation as a session.
DialogTurn(id, user_id, conversation_id, ...)	A turn consists of a user query and the assistant response.
Document(text, meta_data, ...)	A text container with optional metadata and vector representation.
EmbedderOutput(data, model, usage, error, ...)	Container to hold the response from an Embedder datacomponent for a single batch of input.
EmbedderOutputType	alias of EmbedderOutput
Embedding(embedding, index)	Container for a single embedding.
Function(thought, name, args, kwargs, ...)	The data modeling of a function call, including the name and keyword arguments.
FunctionDefinition(class_instance, ...)	The data modeling of a function definition, including the name, description, and parameters.
FunctionExpression(thought, action)	The data modeling of a function expression for a call, including the name and arguments.
FunctionOutput([name, input, parsed_input, ...])	The output of a tool, which could be a function, a class, or a module.
GeneratorOutput([id, data, error, usage, ...])	The output data class for the Generator component.
ModelClientType()	A quick way to access all model clients in the ModelClient module.
ModelType(value[, names, module, qualname, ...])	The type of the model, including Embedder, LLM, Reranker.
RetrieverOutput(id, doc_indices, doc_scores, ...)	Save the output of a single query in retrievers.
StepOutput([step, action, function, observation])	The output of a single step in the agent.
TokenLogProb(token, logprob)	similar to openai.ChatCompletionTokenLogprob
Usage(prompt_tokens, total_tokens)	In sync with OpenAI embedding api spec, same as openai/types/create_embedding_response.py
UserQuery(query_str[, metadata])

class ModelType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

The type of the model, including Embedder, LLM, Reranker. It helps ModelClient identify the model type required to correctly call the model.

EMBEDDER = 1

LLM = 2

RERANKER = 3

IMAGE_GENERATION = 4

UNDEFINED = 5

class ModelClientType

Bases: object

A quick way to access all model clients in the ModelClient module.

From this:

from adalflow.components.model_client import CohereAPIClient, TransformersClient, AnthropicAPIClient, GroqAPIClient, OpenAIClient

model_client = OpenAIClient()

To this:

from adalflow.core.types import ModelClientType

model_client = ModelClientType.OPENAI()

COHERE = <adalflow.utils.lazy_import.LazyImport object>

TRANSFORMERS = <adalflow.utils.lazy_import.LazyImport object>

ANTHROPIC = <adalflow.utils.lazy_import.LazyImport object>

GROQ = <adalflow.utils.lazy_import.LazyImport object>

OPENAI = <adalflow.utils.lazy_import.LazyImport object>

GOOGLE_GENAI = <adalflow.utils.lazy_import.LazyImport object>

OLLAMA = <adalflow.utils.lazy_import.LazyImport object>

get_model_args(model_type: ModelType) → List[str]

Get the required keys in model_kwargs for a specific model type.

note: If your model inference sdk uses different keys, you need to convert them to the standard keys here in their specifc ModelClient.

Parameters:

model_type (ModelType) – The model type

Returns:

The required keys in model_kwargs

Return type:

List[str]

class Embedding(embedding: List[float], index: int | None)

Bases: object

Container for a single embedding.

In sync with api spec, same as openai/types/embedding.py

embedding: List[float]

index: int | None

class Usage(prompt_tokens: int, total_tokens: int)

Bases: object

In sync with OpenAI embedding api spec, same as openai/types/create_embedding_response.py

prompt_tokens: int

total_tokens: int

class EmbedderOutput(data: ~typing.List[~core.types.Embedding] = <factory>, model: str | None = None, usage: ~core.types.Usage | None = None, error: str | None = None, raw_response: ~typing.Any | None = None, input: ~typing.List[str] | None = None)

Bases: DataClass

Container to hold the response from an Embedder datacomponent for a single batch of input.

Data standard for Embedder model output to interact with other components. Batch processing is often available, thus we need a list of Embedding objects.

data: List[Embedding]

model: str | None = None

usage: Usage | None = None

error: str | None = None

raw_response: Any | None = None

input: List[str] | None = None

property length: int

property embedding_dim: int

The dimension of the embedding, assuming all embeddings have the same dimension.

Returns:

The dimension of the embedding, -1 if no embedding is available

Return type:

int

property is_normalized: bool

Check if the embeddings are normalized to unit vectors.

Returns:

True if the embeddings are normalized, False otherwise

Return type:

bool

EmbedderOutputType

alias of EmbedderOutput

class TokenLogProb(token: str, logprob: float)

Bases: object

similar to openai.ChatCompletionTokenLogprob

token: str

logprob: float

class CompletionUsage(completion_tokens: int | None = None, prompt_tokens: int | None = None, total_tokens: int | None = None)

Bases: object

In sync with OpenAI completion usage api spec at openai/types/completion_usage.py

completion_tokens: int | None = None

prompt_tokens: int | None = None

total_tokens: int | None = None

class GeneratorOutput(id: str | None = None, data: T_co = None, error: str | None = None, usage: CompletionUsage | None = None, raw_response: str | None = None, metadata: Dict[str, object] | None = None)

Bases: DataClass, Generic[T_co]

The output data class for the Generator component. We can not control its output 100%, so we use this to track the error_message and allow the raw string output to be passed through.

(1) When model predict and output processors are both without error, we have data as the final output, error as None. (2) When either model predict or output processors have error, we have data as None, error as the error message.

Raw_response will depends on the model predict.

id: str | None = None

data: T_co = None

error: str | None = None

usage: CompletionUsage | None = None

raw_response: str | None = None

metadata: Dict[str, object] | None = None

class RetrieverOutput(id: str = None, doc_indices: ~typing.List[int] = <function required_field>, doc_scores: ~typing.List[float] = None, query: ~core.types.RetrieverQueryType = None, documents: ~typing.List[~core.types.RetrieverDocumentType] = None)

Bases: DataClass

Save the output of a single query in retrievers.

It is up to the subclass of Retriever to specify the type of query and document.

id: str = None

doc_indices() → Callable[[], Any]

A factory function to create a required field in a dataclass. The returned callable raises a TypeError when invoked, indicating a required field was not provided.

Parameters:

name (Optional[str], optional) – The name of the required field. Defaults to None

Returns:

A callable that raises TypeError when called, indicating a missing required field.

Return type:

Callable[[], Any]

Example:

from dataclasses import dataclass
from adalflow.core.base_data_class import required_field, DataClass

@dataclass
class Person(DataClass):
    name: str = field(default=None)
    age: int = field(default_factory=required_field())# allow required field after optional field

doc_scores: List[float] = None

query: RetrieverQueryType = None

documents: List[RetrieverDocumentType] = None

class FunctionDefinition(class_instance: ~typing.Any | None = None, func_name: str = <function required_field>, func_desc: str | None = None, func_parameters: ~typing.Dict[str, object] = <factory>)

Bases: DataClass

The data modeling of a function definition, including the name, description, and parameters.

class_instance: Any | None = None

func_name() → Callable[[], Any]

A factory function to create a required field in a dataclass. The returned callable raises a TypeError when invoked, indicating a required field was not provided.

Parameters:

name (Optional[str], optional) – The name of the required field. Defaults to None

Returns:

A callable that raises TypeError when called, indicating a missing required field.

Return type:

Callable[[], Any]

Example:

from dataclasses import dataclass
from adalflow.core.base_data_class import required_field, DataClass

@dataclass
class Person(DataClass):
    name: str = field(default=None)
    age: int = field(default_factory=required_field())# allow required field after optional field

func_desc: str | None = None

func_parameters: Dict[str, object]

fn_schema_str(type: Literal['json', 'yaml'] = 'json') → str

Get the function definition str to be used in the prompt.

You should also directly use to_json() and to_yaml() to get the schema in JSON or YAML format.

class Function(thought: str | None = None, name: str = '', args: ~typing.List[object] | None = <factory>, kwargs: ~typing.Dict[str, object] | None = <factory>)

Bases: DataClass

The data modeling of a function call, including the name and keyword arguments.

You can use the exclude in to_json() and to_yaml() to exclude the thought field if you do not want to use chain-of-thought pattern.

Example:

# assume the function is added in a context_map
# context_map = {"add": add}

def add(a, b):
    return a + b

# call function add with arguments 1 and 2
fun = Function(name="add", kwargs={"a": 1, "b": 2})
# evaluate the function
result = context_map[fun.name](**fun.kwargs)

# or call with positional arguments
fun = Function(name="add", args=[1, 2])
result = context_map[fun.name](*fun.args)

thought: str | None = None

name: str = ''

args: List[object] | None

kwargs: Dict[str, object] | None

classmethod from_function(func: Callable[[...], Any] | Callable[[...], Awaitable[Any]], thought: str | None = None, *args, **kwargs) → Function

Create a Function object from a function.

Parameters:

fun (Union[Callable[..., Any], AsyncCallable]) – The function to be converted

Returns:

The Function object

Return type:

Function

Usage: 1. Create a Function object from a function call: 2. use to_json() and to_yaml() to get the schema in JSON or YAML format. 3. This will be used as an example in prompt showing LLM how to call the function.

Example:

from adalflow.core.types import Function

def add(a, b):
    return a + b

# create a function call object with positional arguments
fun = Function.from_function(add, thought="Add two numbers", 1, 2)
print(fun)

# output
# Function(thought='Add two numbers', name='add', args=[1, 2])

class FunctionExpression(thought: str = None, action: str = <factory>)

Bases: DataClass

The data modeling of a function expression for a call, including the name and arguments.

Example:

def add(a, b):
    return a + b

# call function add with positional arguments 1 and 2
fun_expr = FunctionExpression(action="add(1, 2)")
# evaluate the expression
result = eval(fun_expr.action)
print(result)
# Output: 3

# call function add with keyword arguments
fun_expr = FunctionExpression(action="add(a=1, b=2)")
result = eval(fun_expr.action)
print(result)
# Output: 3

Why asking LLM to generate function expression (code snippet) for a function call? - It is more efficient/compact to call a function. - It is more flexible.

1. for the full range of Python expressions, including arithmetic operations, nested function calls, and more.

2. allow to pass variables as arguments.

• Ease of parsing using ast module.

The benefits are less failed function calls.

thought: str = None

action: str

classmethod from_function(func: Callable[[...], Any] | Callable[[...], Awaitable[Any]], thought: str | None = None, *args, **kwargs) → FunctionExpression

Create a FunctionExpression object from a function.

Parameters:

fun (Union[Callable[..., Any], AsyncCallable]) – The function to be converted

Returns:

The FunctionExpression object

Return type:

FunctionExpression

Usage: 1. Create a FunctionExpression object from a function call: 2. use to_json() and to_yaml() to get the schema in JSON or YAML format. 3. This will be used as an example in prompt showing LLM how to call the function.

Example:

from adalflow.core.types import FunctionExpression

def add(a, b):
    return a + b

# create an expression for the function call and using keyword arguments
fun_expr = FunctionExpression.from_function(
    add, thought="Add two numbers", a=1, b=2
)
print(fun_expr)

# output
# FunctionExpression(thought='Add two numbers', action='add(a=1, b=2)')

class FunctionOutput(name: str | None = None, input: Function | FunctionExpression | None = None, parsed_input: Function | None = None, output: object | None = None, error: str | None = None)

Bases: DataClass

The output of a tool, which could be a function, a class, or a module.

name: str | None = None

input: Function | FunctionExpression | None = None

parsed_input: Function | None = None

output: object | None = None

error: str | None = None

class StepOutput(step: int = 0, action: T = None, function: Function | None = None, observation: str | None = None)

Bases: DataClass, Generic[T]

The output of a single step in the agent. Suits for serial planning agent such as React

step: int = 0

action: T = None

function: Function | None = None

observation: str | None = None

to_prompt_str() → str

class Document(text: str, meta_data: ~typing.Dict[str, ~typing.Any] | None = None, vector: ~typing.List[float] = <factory>, id: str | None = <factory>, order: int | None = None, score: float | None = None, parent_doc_id: str | ~uuid.UUID | None = None, estimated_num_tokens: int | None = None)

Bases: DataClass

A text container with optional metadata and vector representation.

It is the data structure to support functions like Retriever, DocumentSplitter, and used with LocalDB.

text: str

meta_data: Dict[str, Any] | None = None

vector: List[float]

id: str | None

order: int | None = None

score: float | None = None

parent_doc_id: str | UUID | None = None

estimated_num_tokens: int | None = None

classmethod from_dict(doc: Dict)

Create a Document object from a dictionary.

Example:

doc = Document.from_dict({
    "id": "123",
    "text": "Hello world",
    "meta_data": {"title": "Greeting"}
})

class UserQuery(query_str: str, metadata: Dict[str, Any] | None = None)

Bases: object

query_str: str

metadata: Dict[str, Any] | None = None

class AssistantResponse(response_str: str, metadata: Dict[str, Any] | None = None)

Bases: object

response_str: str

metadata: Dict[str, Any] | None = None

class DialogTurn(id: str = <factory>, user_id: str | None = None, conversation_id: str | None = None, order: int | None = None, user_query: ~core.types.UserQuery | None = None, assistant_response: ~core.types.AssistantResponse | None = None, user_query_timestamp: ~datetime.datetime | None = <factory>, assistant_response_timestamp: ~datetime.datetime | None = <factory>, metadata: ~typing.Dict[str, ~typing.Any] | None = None, vector: ~typing.List[float] | None = None)

Bases: DataClass

A turn consists of a user query and the assistant response.

The dataformat is designed to fit into a relational database, where each turn is a row. Use session_id to group the turns into a dialog session with the order field and user_query_timestamp and assistant_response_timestamp to order the turns.

Parameters:

• id (str) – The unique id of the turn.

• user_id (str, optional) – The unique id of the user.

• session_id (str, optional) – The unique id of the dialog session.

• order (int, optional) – The order of the turn in the dialog session, starts from 0.

• user_query (UserQuery, optional) – The user query in the turn.

• assistant_response (AssistantResponse, optional) – The assistant response in the turn.

• user_query_timestamp (datetime, optional) – The timestamp of the user query.

• assistant_response_timestamp (datetime, optional) – The timestamp of the assistant response.

• metadata (Dict[str, Any], optional) – Additional metadata.

Examples

• User: Hi, how are you?

• Assistant: I’m fine, thank you!

DialogTurn(id=uuid4(), user_query=UserQuery(“Hi, how are you?”), assistant_response=AssistantResponse(“I’m fine, thank you!”))

id: str

user_id: str | None = None

conversation_id: str | None = None

order: int | None = None

user_query: UserQuery | None = None

assistant_response: AssistantResponse | None = None

user_query_timestamp: datetime | None

assistant_response_timestamp: datetime | None

metadata: Dict[str, Any] | None = None

vector: List[float] | None = None

set_user_query(user_query: UserQuery, user_query_timestamp: datetime | None = None)

set_assistant_response(assistant_response: AssistantResponse, assistant_response_timestamp: datetime | None = None)

class Conversation(id: str = <factory>, name: str | None = None, user_id: str | None = None, dialog_turns: ~collections.OrderedDict[int, ~core.types.DialogTurn] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] | None = None, created_at: ~datetime.datetime | None = <factory>, dialog_turns_input: dataclasses.InitVar[typing.Union[collections.OrderedDict[int, core.types.DialogTurn], typing.List[core.types.DialogTurn], NoneType]] = None)

Bases: object

A conversation manages the dialog turns in a whole conversation as a session.

This class is mainly used in-memory for the dialog system/app to manage active conversations. You won’t need this class for past conversations which have already been persisted in a database as a form of record or history.

id: str

name: str | None = None

user_id: str | None = None

dialog_turns: OrderedDict[int, DialogTurn]

metadata: Dict[str, Any] | None = None

created_at: datetime | None

dialog_turns_input: dataclasses.InitVar[Union[collections.OrderedDict[int, core.types.DialogTurn], List[core.types.DialogTurn], NoneType]] = None

get_next_order()

append_dialog_turn(dialog_turn: DialogTurn)

get_dialog_turns() → OrderedDict[int, DialogTurn]

get_chat_history_str() → str

delete_dialog_turn(order: int)

update_dialog_turn(order: int, dialog_turn: DialogTurn)