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 model.

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(func_name, func_desc, ...)

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(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)[source]#

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#
UNDEFINED = 4#
class ModelClientType[source]#

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>#
get_model_args(model_type: ModelType) → List[str][source]#

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)[source]#

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)[source]#

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)[source]#

Bases: DataClass

Container to hold the response from an Embedder model. Only Per-batch.

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)[source]#

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)[source]#

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)[source]#

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(doc_indices: List[int], doc_scores: List[float] | None = None, query: RetrieverQueryType | None = None, documents: List[RetrieverDocumentType] | None = None)[source]#

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.

doc_indices: List[int]#
doc_scores: List[float] | None = None#
query: RetrieverQueryType | None = None#
documents: List[RetrieverDocumentType] | None = None#
class FunctionDefinition(func_name: str, func_desc: str | None = None, func_parameters: ~typing.Dict[str, object] = <factory>)[source]#

Bases: DataClass

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

func_name: str#
func_desc: str | None = None#
func_parameters: Dict[str, object]#
fn_schema_str(type: Literal['json', 'yaml'] = 'json') → str[source]#

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>)[source]#

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#
class FunctionExpression(thought: str | None = None, action: str = <factory>)[source]#

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 = None#
action: str#
classmethod from_function(func: Callable[[...], Any] | Callable[[...], Awaitable[Any]], thought: str | None = None, *args, **kwargs) → FunctionExpression[source]#

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)[source]#

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)[source]#

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#
classmethod with_action_type(action_type: Type[T]) → Type[StepOutput[T]][source]#

Create a new StepOutput class with the specified action type.

Use this if you want to create schema for StepOutput with a specific action type.

Parameters:

action_type (Type[T]) – The type to set for the action attribute.

Returns:

A new subclass of StepOutput with the specified action type.

Return type:

Type[StepOutput[T]]

Example:

from adalflow.core.types import StepOutput, FunctionExpression

StepOutputWithFunctionExpression = StepOutput.with_action_type(FunctionExpression)
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)[source]#

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)[source]#

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)[source]#

Bases: object

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

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)[source]#

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)[source]#
set_assistant_response(assistant_response: AssistantResponse, assistant_response_timestamp: datetime | None = None)[source]#
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)[source]#

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()[source]#
append_dialog_turn(dialog_turn: DialogTurn)[source]#
get_dialog_turns() → OrderedDict[int, DialogTurn][source]#
get_chat_history_str() → str[source]#
delete_dialog_turn(order: int)[source]#
update_dialog_turn(order: int, dialog_turn: DialogTurn)[source]#
On this page