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.structured_tools.python_tool import PythonTool
from amsdal_ml.ml_models.openai.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-5.1'),
    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¶

The agent receives the question and a list of available tools with descriptions
At each step, the LLM generates a Thought (reasoning) and an Action (tool call with arguments)
The tool is executed and the Observation (result) is added to the agent's scratchpad
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.structured_tools.python_tool import PythonTool
from amsdal_ml.ml_models.openai.openai_model import OpenAIModel

agent = FunctionalCallingAgent(
    model=OpenAIModel(model_name='gpt-5.1'),
    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

Native MCP Servers¶

FunctionalCallingAgent can also connect to MCP servers natively (in addition to client-side tools) by passing mcp_servers, and accepts a custom adapter:

from amsdal_ml.ml_models.primitives import MCPServerConfig

agent = FunctionalCallingAgent(
    model=OpenAIModel(model_name='gpt-5.1'),
    mcp_servers=MCPServerConfig(...),  # one config or a list
)

DefaultQAAgent is async-only — call arun(). Its synchronous run() raises NotImplementedError. FunctionalCallingAgent also supports a synchronous run().

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
`message`	`ChatMessage \\| None`	The full assistant message (populated by `FunctionalCallingAgent`)
`used_tools`	`list[str]`	Names of tools that were called
`citations`	`list[dict[str, Any]]`	Source citations (if available)

Memory & Attachments¶

Both agents maintain their conversation history using a MessageBuffer, which stores a list of ChatMessage objects. The ChatMessage representation is agnostic to specific LLM provider formats and fully supports storing the history of file attachments.

Using ChatMessage and MessageBuffer¶

You can manually manage memory by creating ChatMessage objects and appending them to a MessageBuffer. This is particularly useful for injecting system instructions or pre-loading conversational context before the agent runs.

from amsdal_ml.ml_models.primitives import ChatMessage, MessageRole
from amsdal_ml.agents.memory.buffer import MessageBuffer

# Initialize the buffer and manually add ChatMessages
buffer = MessageBuffer()

# Add a system prompt
buffer.append(ChatMessage(
    role=MessageRole.SYSTEM,
    content="You are a helpful, concise assistant."
))

# You can also use utility methods like `add_user` and `add_assistant`
buffer.add_user("What is the capital of France?")
buffer.add_assistant("Paris")

Passing File Attachments¶

ChatMessage also fully supports file attachments. The following example shows how to upload a file using the OpenAI API and attach it to a user message within the buffer:

from openai import AsyncOpenAI
from amsdal_ml.fileio.openai_loader import OpenAIFileLoader
from amsdal_ml.fileio.base_loader import FileItem

# Upload a file via OpenAI and get a FileAttachment
client = AsyncOpenAI()
loader = OpenAIFileLoader(client)

with open('document.pdf', 'rb') as f:
    attachment = await loader.load(FileItem(file=f, filename='document.pdf'))

# Add the attachment to a new user message
buffer.add_user("Summarize this document", attachments=[attachment])

# You can pass the pre-populated buffer history to the agent:
output = await agent.arun(
    'Please summarize the uploaded files.',
    history=buffer.snapshot(),
)