Agent

datapizza.agents.agent.Agent

__init__

__init__(
    name=None,
    client=None,
    *,
    system_prompt=None,
    tools=None,
    max_steps=None,
    terminate_on_text=True,
    stateless=True,
    gen_args=None,
    memory=None,
    stream=None,
    can_call=None,
    logger=None,
    planning_interval=0,
    planning_prompt=PLANNING_PROMT,
)

Initialize the agent.

Parameters:

Name	Type	Description	Default
name	str	The name of the agent. Defaults to None.	None
client	Client	The client to use for the agent. Defaults to None.	None
system_prompt	str	The system prompt to use for the agent. Defaults to None.	None
tools	list[Tool]	A list of tools to use with the agent. Defaults to None.	None
max_steps	int	The maximum number of steps to execute. Defaults to None.	None
terminate_on_text	bool	Whether to terminate the agent on text. Defaults to True.	True
stateless	bool	Whether to use stateless execution. Defaults to True.	True
gen_args	dict[str, Any]	Additional arguments to pass to the agent's execution. Defaults to None.	None
memory	Memory	The memory to use for the agent. Defaults to None.	None
stream	bool	Whether to stream the agent's execution. Defaults to None.	None
can_call	list[Agent]	A list of agents that can call the agent. Defaults to None.	None
logger	AgentLogger	The logger to use for the agent. Defaults to None.	None
planning_interval	int	The planning interval to use for the agent. Defaults to 0.	0
planning_prompt	str	The planning prompt to use for the agent planning steps. Defaults to PLANNING_PROMT.	PLANNING_PROMT

a_run async

a_run(task_input, tool_choice='auto', **gen_kwargs)

Run the agent on a task input asynchronously.

Parameters:

Name	Type	Description	Default
task_input	str	The input text/prompt to send to the model	required
tool_choice	Literal['auto', 'required', 'none', 'required_first'] | list[str]	Controls which tool to use ("auto" by default)	'auto'
**gen_kwargs		Additional keyword arguments to pass to the agent's execution	{}

Returns:

Type	Description
StepResult | None	The final result of the agent's execution

a_stream_invoke async

a_stream_invoke(
    task_input, tool_choice="auto", **gen_kwargs
)

Stream the agent's execution asynchronously, yielding intermediate steps and final result.

Parameters:

Name	Type	Description	Default
task_input	str	The input text/prompt to send to the model	required
tool_choice	Literal['auto', 'required', 'none', 'required_first'] | list[str]	Controls which tool to use ("auto" by default)	'auto'
**gen_kwargs		Additional keyword arguments to pass to the agent's execution	{}

Yields:

Type	Description
AsyncGenerator[ClientResponse | StepResult | Plan | None]	The intermediate steps and final result of the agent's execution

run

run(task_input, tool_choice='auto', **gen_kwargs)

Run the agent on a task input.

Parameters:

Name	Type	Description	Default
task_input	str	The input text/prompt to send to the model	required
tool_choice	Literal['auto', 'required', 'none', 'required_first'] | list[str]	Controls which tool to use ("auto" by default)	'auto'
**gen_kwargs		Additional keyword arguments to pass to the agent's execution	{}

Returns:

Type	Description
StepResult | None	The final result of the agent's execution

stream_invoke

stream_invoke(task_input, tool_choice='auto', **gen_kwargs)

Stream the agent's execution, yielding intermediate steps and final result.

Parameters:

Name	Type	Description	Default
task_input	str	The input text/prompt to send to the model	required
tool_choice	Literal['auto', 'required', 'none', 'required_first'] | list[str]	Controls which tool to use ("auto" by default)	'auto'
**gen_kwargs		Additional keyword arguments to pass to the agent's execution	{}

Yields:

Type	Description
ClientResponse | StepResult | Plan | None	The intermediate steps and final result of the agent's execution