evaluate()

evaluate() is the entry point for running an agent inference. It drives the recursive event loop until an AssistantInferenceResponse is produced.

Signature

async def evaluate(
    controller: AssistantInferenceController,
    event: BaseEvent,
    registry: AlquimiaRegistry,
    tool_executor: Callable[[str, dict[str, Any]], Any] | None = None,
    _depth: int = 0,
) -> AssistantInferenceResponse

Parameters

Parameter	Type	Description
controller	AssistantInferenceController	The controller instance for this inference run. Carries context metadata (task_id, session_id, user_id, etc.)
event	BaseEvent	The initial event to process. Typically an AssistantInference command
registry	AlquimiaRegistry	The agent registry. Used to load agent configs and resolve A2A agents
tool_executor	Callable	Optional callback for client-side tool execution. Called with (tool_name, parameters)
_depth	int	Internal recursion depth counter. Do not set manually

Returns

AssistantInferenceResponse — the final response from the agent.

class AssistantInferenceResponse(BaseResponse):
    result: AIMessage          # The LLM's final message
    conversation: Conversation # Updated conversation history

Raises

Exception	Condition
RecursionError	_depth exceeds MAX_EVALUATE_DEPTH (10,000) — the evaluate() loop itself
A2AMaxDepthReached	Agent-to-agent delegation depth exceeds A2A_MAX_DEPTH (default 5)
MaxIterationsReachedException	A tool exceeded max_steps in the evaluation strategy
MaxIterationVolumeReachedException	A tool exceeded max_concurrent_tools in the evaluation strategy

See Exceptions Reference for full details on all exception types.

Usage

Basic inference

import asyncio
from alquimia.core import evaluate, get_assistant_config, get_local_session
from alquimia.core.assistant import AssistantInferenceController
from alquimia.core.base import CommonAttributes
from alquimia.core.command import AssistantInference
from alquimia.core.observability import setup_observability
from alquimia.registry.registry import AlquimiaRegistry

# Enable metrics — call once at startup, not per-request
setup_observability()

async def run_agent(query: str, assistant_id: str) -> str:
    registry = AlquimiaRegistry()
    registry.load({"agentspace_id": "default"})

    config = get_assistant_config(registry, assistant_id)
    conversation = get_local_session()

    controller = AssistantInferenceController(
        context_metadata=CommonAttributes(
            assistant_id=assistant_id,
            agentspace_id="default",
            user_id="user-1",
            session_id=conversation.session_id,
        )
    )

    command = AssistantInference(
        query=query,
        conversation=conversation,
        parameters=config,
    )

    response = await evaluate(controller, command, registry)
    return response.result.content

result = asyncio.run(run_agent("What is 2 + 2?", "my-agent"))
print(result)  # "4"

With client-side tool execution

Use tool_executor when the agent has ClientToolExecution tools that should be handled by your application:

async def my_tool_executor(tool_name: str, parameters: dict) -> Any:
    if tool_name == "get_weather":
        return {"temperature": 22, "condition": "sunny"}
    raise ValueError(f"Unknown tool: {tool_name}")

response = await evaluate(
    controller,
    command,
    registry,
    tool_executor=my_tool_executor,
)

With a persistent session

from alquimia.core import get_local_session, save_local_session

# Load existing session (or create new)
conversation = get_local_session(session_id="user-1-session")

# Run inference
response = await evaluate(controller, command, registry)

# Session is auto-saved by ContextPersistence events.
# You can also save manually:
save_local_session("user-1-session")

Helper functions

get_assistant_config(registry, assistant_id)

Load and validate an AssistantConfig from the registry.

from alquimia.core import get_assistant_config

config = get_assistant_config(registry, "my-agent")
# Returns: AssistantConfig

get_local_session(session_id=None)

Load a session from memory or disk, creating a new one if it doesn’t exist.

from alquimia.core import get_local_session

# New session
conversation = get_local_session()

# Existing session
conversation = get_local_session(session_id="abc-123")

flush_local_session(session_id)

Delete a persisted session from disk and memory.

from alquimia.core import flush_local_session

flush_local_session("abc-123")

make_from_yaml(path)

Load a YAML agent configuration file with ${ENV_VAR} substitution.

from alquimia.core import make_from_yaml

config_dict = make_from_yaml("./agents/my-agent.yaml")

Constants

Constant	Value	Description
MAX_EVALUATE_DEPTH	10_000	Maximum recursion depth for the evaluate() loop before RecursionError

Related pages

• Architecture — how evaluate() fits in the execution model
• Event Model — the events that evaluate() dispatches
• AssistantConfig & Profile — the config object passed to evaluate()
• Exceptions Reference — all exceptions raised by evaluate()
• Core SDK Observability — setup_observability() and metrics