AI Agents

amsdal_ml includes two agent implementations that combine LLM reasoning with tool execution to answer complex questions.

DefaultQAAgent (ReAct)

The ReAct agent follows a Reasoning + Acting loop: it thinks about the question, decides which tool to call, observes the result, and repeats until it has an answer.

from amsdal_ml.agents.default_qa_agent import DefaultQAAgent
from amsdal_ml.agents.python_tool import PythonTool
from amsdal_ml.ml_models.openai_model import OpenAIModel

async def search_customers(query: str) -> str:
    """Search the customer database."""
    # your search logic
    return results

agent = DefaultQAAgent(
    model=OpenAIModel(model_name='gpt-4o'),
    tools=[PythonTool(search_customers, name='search_customers', description='Search the customer database.')],
    max_steps=6,
)

output = await agent.arun('Find all customers with overdue invoices')
print(output.answer)
print(output.used_tools)  # list of tools that were called

Configuration

Parameter	Default	Description
max_steps	6	Maximum reasoning steps before stopping
per_call_timeout	20.0	Timeout per tool call (seconds)
on_parse_error	RAISE	What to do on parse errors: RAISE or RETRY
enable_stop_guard	True	Enable guard against premature stopping

How It Works

1. The agent receives the question and a list of available tools with descriptions
2. At each step, the LLM generates a Thought (reasoning) and an Action (tool call with arguments)
3. The tool is executed and the Observation (result) is added to the agent's scratchpad
4. Steps repeat until the LLM produces a Final Answer or max_steps is reached

FunctionalCallingAgent

Uses the LLM's native function-calling API (e.g., OpenAI tool_calls) instead of text-based parsing. More reliable for structured tool invocation.

from amsdal_ml.agents.functional_calling_agent import FunctionalCallingAgent
from amsdal_ml.agents.python_tool import PythonTool
from amsdal_ml.ml_models.openai_model import OpenAIModel

agent = FunctionalCallingAgent(
    model=OpenAIModel(model_name='gpt-4o'),
    tools=[PythonTool(search_customers, name='search_customers', description='Search the customer database.')],
    max_steps=10,
)

output = await agent.arun('How many active customers do we have?')
print(output.answer)

Key Differences from ReAct

Aspect	DefaultQAAgent (ReAct)	FunctionalCallingAgent
Tool invocation	Regex-parsed from LLM text	Native function calling API
Reliability	May need retries on parse errors	More reliable parsing
Message history	Scratchpad-based	Full message history
Streaming	Step-by-step streaming	Buffers full result, yields once
Multi-tool	One tool per step	Multiple tools per step

Streaming

Both agents support streaming via astream():

async for chunk in agent.astream('Summarize customer activity'):
    print(chunk, end='', flush=True)

AgentOutput

Both agents return an AgentOutput:

Field	Type	Description
answer	str	The agent's final answer
used_tools	list[str]	Names of tools that were called
citations	list[dict[str, Any]]	Source citations (if available)

Attachments

Both agents support file attachments:

from amsdal_ml.fileio.base_loader import FileAttachment

attachment = FileAttachment(
    type='file_id',
    content='file-abc123',
    mime_type='application/pdf',
)

output = await agent.arun(
    'Summarize this document',
    attachments=[attachment],
)