Methods

__init__(template=None, memory=None, **kwargs)

Creates a new Agent.

Parameters:

Name	Type	Description	Default
template		optional prompt jinja template, must include {{ text }} and {{ memory }} placeholders	None
memory		number of prior outputs to keep as "memory", defaults to None for no memory	None
kwargs		arguments to pass to the underlying Agent backend and LLM pipeline instance	{}

Source code in txtai/agent/base.py

def __init__(self, template=None, memory=None, **kwargs):
    """
    Creates a new Agent.

    Args:
        template: optional prompt jinja template, must include {{ text }} and {{ memory }} placeholders
        memory: number of prior outputs to keep as "memory", defaults to None for no memory
        kwargs: arguments to pass to the underlying Agent backend and LLM pipeline instance
    """

    # Ensure backwards compatibility
    if "max_iterations" in kwargs:
        kwargs["max_steps"] = kwargs.pop("max_iterations")

    # Custom instructions
    if "instructions" in kwargs:
        kwargs["instructions"] = self.instructions(kwargs)

    # Create agent process runner
    self.process = ProcessFactory.create(kwargs)

    # Tools dictionary
    self.tools = self.process.tools

    # Agent memory
    self.memory = deque(maxlen=memory) if memory else None
    self.template = template

__call__(text, maxlength=8192, stream=False, reset=False, **kwargs)

Runs an agent loop.

Parameters:

Name	Type	Description	Default
text		instructions to run	required
maxlength		maximum sequence length	8192
stream		stream response if True, defaults to False	False
reset		clears previously stored memory if True, defaults to False	False
kwargs		additional keyword arguments	{}

Returns:

Type	Description
	result

Source code in txtai/agent/base.py

def __call__(self, text, maxlength=8192, stream=False, reset=False, **kwargs):
    """
    Runs an agent loop.

    Args:
        text: instructions to run
        maxlength: maximum sequence length
        stream: stream response if True, defaults to False
        reset: clears previously stored memory if True, defaults to False
        kwargs: additional keyword arguments

    Returns:
        result
    """

    # Process parameters
    self.process.model.parameters(maxlength)

    # Clear memory, if reset flag set
    if reset and self.memory:
        self.memory.clear()

    # Run agent loop
    output = self.process.run(self.prompt(text), stream=stream, **kwargs)

    # Add output to memory, if necessary
    if self.memory is not None:
        self.memory.append((text, output))

    return output