Skip to main content

​
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