Camel.data collectors.base

​

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
):

• 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]):

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

​

BaseDataCollector

class BaseDataCollector(ABC):

​

init

def __init__(self):

​

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
):

• 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)

​

record

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

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

​

start

def start(self):

​

stop

def stop(self):

​

recording

def recording(self):

​

reset

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

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

​

convert

def convert(self):

​

llm_convert

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

​

get_agent_history

def get_agent_history(self, name: str):

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