​
CollectorData

class CollectorData:

​
init

def __init__(
    self,
    id: UUID,
    name: str,
    role: Literal['user', 'assistant', 'system', 'tool'],
    message: Optional[str] = None,
    function_call: Optional[Dict[str, Any]] = None
):

Create a data item store information about a message. Used by the data collector.

Parameters:

  • id (UUID): The id of the message.
  • name (str): The name of the agent.
  • role (Literal["user", "assistant", "system", "function"]): The role of the message.
  • message (Optional[str], optional): The message. (default: :obj:None)
  • function_call (Optional[Dict[str, Any]], optional): The function call. (default: :obj:None)

​
from_context

def from_context(name, context: Dict[str, Any]):

Create a data collector from a context.

Parameters:

  • name (str): The name of the agent.
  • context (Dict[str, Any]): The context.

Returns:

CollectorData: The data collector.

​
BaseDataCollector

class BaseDataCollector(ABC):

Base class for data collectors.

​
init

def __init__(self):

Create a data collector.

​
step

def step(
    self,
    role: Literal['user', 'assistant', 'system', 'tool'],
    name: Optional[str] = None,
    message: Optional[str] = None,
    function_call: Optional[Dict[str, Any]] = None
):

Record a message.

Parameters:

  • role (Literal["user", "assistant", "system", "tool"]): The role of the message.
  • name (Optional[str], optional): The name of the agent. (default: :obj:None)
  • message (Optional[str], optional): The message to record. (default: :obj:None)
  • function_call (Optional[Dict[str, Any]], optional): The function call to record. (default: :obj:None)

Returns:

Self: The data collector.

​
record

def record(self, agent: Union[List[ChatAgent], ChatAgent]):

Record agents.

Parameters:

  • agent (Union[List[ChatAgent], ChatAgent]): The agent(s) to inject.

​
start

def start(self):

Start recording.

​
stop

def stop(self):

Stop recording.

​
recording

def recording(self):

Whether the collector is recording.

​
reset

def reset(self, reset_agents: bool = True):

Reset the collector.

Parameters:

  • reset_agents (bool, optional): Whether to reset the agents. Defaults to True.

​
convert

def convert(self):

Convert the collected data.

​
llm_convert

def llm_convert(self, converter: Any, prompt: Optional[str] = None):

Convert the collected data.

​
get_agent_history

def get_agent_history(self, name: str):

Get the message history of an agent.

Parameters:

  • name (str): The name of the agent.

Returns:

List[CollectorData]: The message history of the agent