Camel.agents.chat agent

​

StreamContentAccumulator

class StreamContentAccumulator:

​

init

def __init__(self):

​

set_base_content

def set_base_content(self, content: str):

​

add_streaming_content

def add_streaming_content(self, new_content: str):

​

add_tool_status

def add_tool_status(self, status_message: str):

​

get_full_content

def get_full_content(self):

​

get_content_with_new_status

def get_content_with_new_status(self, status_message: str):

​

reset_streaming_content

def reset_streaming_content(self):

​

StreamingChatAgentResponse

class StreamingChatAgentResponse:

​

init

def __init__(self, generator: Generator[ChatAgentResponse, None, None]):

​

_ensure_latest_response

def _ensure_latest_response(self):

​

msgs

def msgs(self):

​

terminated

def terminated(self):

​

info

def info(self):

​

msg

def msg(self):

​

iter

def __iter__(self):

​

getattr

def __getattr__(self, name):

​

AsyncStreamingChatAgentResponse

class AsyncStreamingChatAgentResponse:

​

init

def __init__(self, async_generator: AsyncGenerator[ChatAgentResponse, None]):

​

await

def __await__(self):

​

aiter

def __aiter__(self):

​

ChatAgent

class ChatAgent(BaseAgent):

• system_message (Union[BaseMessage, str], optional): The system message for the chat agent. (default: :obj:None) model (Union[BaseModelBackend, Tuple[str, str], str, ModelType, Tuple[ModelPlatformType, ModelType], List[BaseModelBackend], List[str], List[ModelType], List[Tuple[str, str]], List[Tuple[ModelPlatformType, ModelType]]], optional): The model backend(s) to use. Can be a single instance, a specification (string, enum, tuple), or a list of instances or specifications to be managed by ModelManager. If a list of specifications (not BaseModelBackend instances) is provided, they will be instantiated using ModelFactory. (default: :obj:ModelPlatformType.DEFAULT with ModelType.DEFAULT)
• memory (AgentMemory, optional): The agent memory for managing chat messages. If None, a :obj:ChatHistoryMemory will be used. (default: :obj:None)
• message_window_size (int, optional): The maximum number of previous messages to include in the context window. If None, no windowing is performed. (default: :obj:None)
• token_limit (int, optional): The maximum number of tokens in a context. The context will be automatically pruned to fulfill the limitation. If None, it will be set according to the backend model. (default: :obj:None)
• output_language (str, optional): The language to be output by the agent. (default: :obj:None)
• tools (Optional[List[Union[FunctionTool, Callable]]], optional): List of available :obj:FunctionTool or :obj:Callable. (default: :obj:None) toolkits_to_register_agent (Optional[List[RegisteredAgentToolkit]], optional): List of toolkit instances that inherit from :obj:RegisteredAgentToolkit. The agent will register itself with these toolkits, allowing them to access the agent instance. Note: This does NOT add the toolkit’s tools to the agent. To use tools from these toolkits, pass them explicitly via the tools parameter. (default: :obj:None) external_tools (Optional[List[Union[FunctionTool, Callable, Dict[str, Any]]]], optional): List of external tools (:obj:FunctionTool or :obj:Callable or :obj:Dict[str, Any]) bind to one chat agent. When these tools are called, the agent will directly return the request instead of processing it. (default: :obj:None)
• response_terminators (List[ResponseTerminator], optional): List of :obj:ResponseTerminator bind to one chat agent. (default: :obj:None)
• scheduling_strategy (str): name of function that defines how to select the next model in ModelManager. (default: :str:round_robin)
• max_iteration (Optional[int], optional): Maximum number of model calling iterations allowed per step. If None (default), there’s no explicit limit. If 1, it performs a single model call. If N > 1, it allows up to N model calls. (default: :obj:None)
• agent_id (str, optional): The ID of the agent. If not provided, a random UUID will be generated. (default: :obj:None)
• stop_event (Optional[threading.Event], optional): Event to signal termination of the agent’s operation. When set, the agent will terminate its execution. (default: :obj:None)
• tool_execution_timeout (Optional[float], optional): Timeout for individual tool execution. If None, wait indefinitely.
• mask_tool_output (Optional[bool]): Whether to return a sanitized placeholder instead of the raw tool output. (default: :obj:False)
• pause_event (Optional[asyncio.Event]): Event to signal pause of the agent’s operation. When clear, the agent will pause its execution. (default: :obj:None)
• prune_tool_calls_from_memory (bool): Whether to clean tool call messages from memory after response generation to save token usage. When enabled, removes FUNCTION/TOOL role messages and ASSISTANT messages with tool_calls after each step. (default: :obj:False)
• retry_attempts (int, optional): Maximum number of retry attempts for rate limit errors. (default: :obj:3)
• retry_delay (float, optional): Initial delay in seconds between retries. Uses exponential backoff. (default: :obj:1.0)
• step_timeout (Optional[float], optional): Timeout in seconds for the entire step operation. If None, no timeout is applied. (default: :obj:None)
• stream_accumulate (bool, optional): When True, partial streaming updates return accumulated content (current behavior). When False, partial updates return only the incremental delta. (default: :obj:True)

​

init

def __init__(
    self,
    system_message: Optional[Union[BaseMessage, str]] = None,
    model: Optional[Union[BaseModelBackend, ModelManager, Tuple[str, str], str, ModelType, Tuple[ModelPlatformType, ModelType], List[BaseModelBackend], List[str], List[ModelType], List[Tuple[str, str]], List[Tuple[ModelPlatformType, ModelType]]]] = None,
    memory: Optional[AgentMemory] = None,
    message_window_size: Optional[int] = None,
    token_limit: Optional[int] = None,
    output_language: Optional[str] = None,
    tools: Optional[List[Union[FunctionTool, Callable]]] = None,
    toolkits_to_register_agent: Optional[List[RegisteredAgentToolkit]] = None,
    external_tools: Optional[List[Union[FunctionTool, Callable, Dict[str, Any]]]] = None,
    response_terminators: Optional[List[ResponseTerminator]] = None,
    scheduling_strategy: str = 'round_robin',
    max_iteration: Optional[int] = None,
    agent_id: Optional[str] = None,
    stop_event: Optional[threading.Event] = None,
    tool_execution_timeout: Optional[float] = None,
    mask_tool_output: bool = False,
    pause_event: Optional[asyncio.Event] = None,
    prune_tool_calls_from_memory: bool = False,
    retry_attempts: int = 3,
    retry_delay: float = 1.0,
    step_timeout: Optional[float] = None,
    stream_accumulate: bool = True
):

​

reset

def reset(self):

​

_resolve_models

def _resolve_models(
    self,
    model: Optional[Union[BaseModelBackend, Tuple[str, str], str, ModelType, Tuple[ModelPlatformType, ModelType], List[BaseModelBackend], List[str], List[ModelType], List[Tuple[str, str]], List[Tuple[ModelPlatformType, ModelType]]]]
):

• model: Model specification in various formats including single model, list of models, or model type specifications.

​

_resolve_model_list

def _resolve_model_list(self, model_list: list):

• model_list (list): List of model specifications in various formats.

​

system_message

def system_message(self):

​

tool_dict

def tool_dict(self):

​

output_language

def output_language(self):

​

output_language

def output_language(self, value: str):

​

memory

def memory(self):

​

memory

def memory(self, value: AgentMemory):

• value (AgentMemory): The new agent memory to use.

​

_get_full_tool_schemas

def _get_full_tool_schemas(self):

​

_get_external_tool_names

def _get_external_tool_names(self):

​

add_tool

def add_tool(self, tool: Union[FunctionTool, Callable]):

​

add_tools

def add_tools(self, tools: List[Union[FunctionTool, Callable]]):

​

add_external_tool

def add_external_tool(self, tool: Union[FunctionTool, Callable, Dict[str, Any]]):

​

remove_tool

def remove_tool(self, tool_name: str):

• tool_name (str): The name of the tool to remove.

​

remove_tools

def remove_tools(self, tool_names: List[str]):

​

remove_external_tool

def remove_external_tool(self, tool_name: str):

• tool_name (str): The name of the tool to remove.

​

update_memory

def update_memory(
    self,
    message: BaseMessage,
    role: OpenAIBackendRole,
    timestamp: Optional[float] = None
):

• message (BaseMessage): The new message to add to the stored messages.
• role (OpenAIBackendRole): The backend role type.
• timestamp (Optional[float], optional): Custom timestamp for the memory record. If None, the current time will be used. (default: :obj:None) (default: obj:None)

​

load_memory

def load_memory(self, memory: AgentMemory):

• memory (AgentMemory): The memory to load into the agent.

​

load_memory_from_path

def load_memory_from_path(self, path: str):

• path (str): The file path to a JSON memory file that uses JsonStorage.

​

save_memory

def save_memory(self, path: str):

• path (str): Target file path to store JSON data.

​

summarize

def summarize(
    self,
    filename: Optional[str] = None,
    summary_prompt: Optional[str] = None,
    working_directory: Optional[Union[str, Path]] = None
):

• filename (Optional[str]): The base filename (without extension) to use for the markdown file. Defaults to a timestamped name when not provided.
• summary_prompt (Optional[str]): Custom prompt for the summarizer. When omitted, a default prompt highlighting key decisions, action items, and open questions is used.
• working_directory (Optional[str|Path]): Optional directory to save the markdown summary file. If provided, overrides the default directory used by ContextUtility.

​

clear_memory

def clear_memory(self):

​

_generate_system_message_for_output_language

def _generate_system_message_for_output_language(self):

​

init_messages

def init_messages(self):

​

record_message

def record_message(self, message: BaseMessage):

• message (BaseMessage): An external message to be recorded in the memory.

​

_try_format_message

def _try_format_message(self, message: BaseMessage, response_format: Type[BaseModel]):

​

_check_tools_strict_compatibility

def _check_tools_strict_compatibility(self):

​

_convert_response_format_to_prompt

def _convert_response_format_to_prompt(self, response_format: Type[BaseModel]):

• response_format (Type[BaseModel]): The Pydantic model class.

​

_handle_response_format_with_non_strict_tools

def _handle_response_format_with_non_strict_tools(
    self,
    input_message: Union[BaseMessage, str],
    response_format: Optional[Type[BaseModel]] = None
):

• input_message: The original input message.
• response_format: The requested response format.

​

_is_called_from_registered_toolkit

def _is_called_from_registered_toolkit(self):

​

_apply_prompt_based_parsing

def _apply_prompt_based_parsing(
    self,
    response: ModelResponse,
    original_response_format: Type[BaseModel]
):

• response: The model response to parse.
• original_response_format: The original response format class.

​

_format_response_if_needed

def _format_response_if_needed(
    self,
    response: ModelResponse,
    response_format: Optional[Type[BaseModel]] = None
):

1. The response format is None (not provided)
2. The response is empty

​

step

def step(
    self,
    input_message: Union[BaseMessage, str],
    response_format: Optional[Type[BaseModel]] = None
):

• input_message (Union[BaseMessage, str]): The input message for the agent. If provided as a BaseMessage, the role is adjusted to user to indicate an external message.
• response_format (Optional[Type[BaseModel]], optional): A Pydantic model defining the expected structure of the response. Used to generate a structured response if provided. (default: :obj:None)

​

_step_impl

def _step_impl(
    self,
    input_message: Union[BaseMessage, str],
    response_format: Optional[Type[BaseModel]] = None
):

​

chat_history

def chat_history(self):

​

_create_token_usage_tracker

def _create_token_usage_tracker(self):

​

_update_token_usage_tracker

def _update_token_usage_tracker(self, tracker: Dict[str, int], usage_dict: Dict[str, int]):

• tracker (Dict[str, int]): The token usage tracker to update.
• usage_dict (Dict[str, int]): The usage dictionary with new values.

​

_convert_to_chatagent_response

def _convert_to_chatagent_response(
    self,
    response: ModelResponse,
    tool_call_records: List[ToolCallingRecord],
    num_tokens: int,
    external_tool_call_requests: Optional[List[ToolCallRequest]],
    step_api_prompt_tokens: int = 0,
    step_api_completion_tokens: int = 0,
    step_api_total_tokens: int = 0
):

​

_record_final_output

def _record_final_output(self, output_messages: List[BaseMessage]):

​

_get_model_response

def _get_model_response(
    self,
    openai_messages: List[OpenAIMessage],
    num_tokens: int,
    current_iteration: int = 0,
    response_format: Optional[Type[BaseModel]] = None,
    tool_schemas: Optional[List[Dict[str, Any]]] = None,
    prev_num_openai_messages: int = 0
):

​

_sanitize_messages_for_logging

def _sanitize_messages_for_logging(self, messages, prev_num_openai_messages: int):

• messages (List[OpenAIMessage]): The OpenAI messages to sanitize.
• prev_num_openai_messages (int): The number of openai messages logged in the previous iteration.

​

_step_get_info

def _step_get_info(
    self,
    output_messages: List[BaseMessage],
    finish_reasons: List[str],
    usage_dict: Dict[str, int],
    response_id: str,
    tool_calls: List[ToolCallingRecord],
    num_tokens: int,
    external_tool_call_requests: Optional[List[ToolCallRequest]] = None
):

• output_messages (List[BaseMessage]): The messages generated in this step.
• finish_reasons (List[str]): The reasons for finishing the generation for each message.
• usage_dict (Dict[str, int]): Dictionary containing token usage information.
• response_id (str): The ID of the response from the model.
• tool_calls (List[ToolCallingRecord]): Records of function calls made during this step.
• num_tokens (int): The number of tokens used in this step.
• external_tool_call_request (Optional[ToolCallRequest]): The request for external tool call.

​

_handle_batch_response

def _handle_batch_response(self, response: ChatCompletion):

• response (ChatCompletion): Model response.

​

_step_terminate

def _step_terminate(
    self,
    num_tokens: int,
    tool_calls: List[ToolCallingRecord],
    termination_reason: str
):

• num_tokens (int): Number of tokens in the messages.
• tool_calls (List[ToolCallingRecord]): List of information objects of functions called in the current step.
• termination_reason (str): String describing the reason for termination.

​

_execute_tool

def _execute_tool(self, tool_call_request: ToolCallRequest):

• tool_call_request (_ToolCallRequest): The tool call request.

​

_record_tool_calling

def _record_tool_calling(
    self,
    func_name: str,
    args: Dict[str, Any],
    result: Any,
    tool_call_id: str,
    mask_output: bool = False
):

• func_name (str): The name of the tool function called.
• args (Dict[str, Any]): The arguments passed to the tool.
• result (Any): The result returned by the tool execution.
• tool_call_id (str): A unique identifier for the tool call.
• mask_output (bool, optional): Whether to return a sanitized placeholder instead of the raw tool output. (default: :obj:False)

​

_stream

def _stream(
    self,
    input_message: Union[BaseMessage, str],
    response_format: Optional[Type[BaseModel]] = None
):

• input_message (Union[BaseMessage, str]): The input message for the agent.
• response_format (Optional[Type[BaseModel]], optional): A Pydantic model defining the expected structure of the response.
• Yields:
• ChatAgentResponse: Intermediate responses containing partial content, tool calls, and other information as they become available.

​

_get_token_count

def _get_token_count(self, content: str):

​

_stream_response

def _stream_response(
    self,
    openai_messages: List[OpenAIMessage],
    num_tokens: int,
    response_format: Optional[Type[BaseModel]] = None
):

​

_process_stream_chunks_with_accumulator

def _process_stream_chunks_with_accumulator(
    self,
    stream: Stream[ChatCompletionChunk],
    content_accumulator: StreamContentAccumulator,
    accumulated_tool_calls: Dict[str, Any],
    tool_call_records: List[ToolCallingRecord],
    step_token_usage: Dict[str, int],
    response_format: Optional[Type[BaseModel]] = None
):

​

_accumulate_tool_calls

def _accumulate_tool_calls(
    self,
    tool_call_deltas: List[Any],
    accumulated_tool_calls: Dict[str, Any]
):

• tool_call_deltas (List[Any]): List of tool call deltas.
• accumulated_tool_calls (Dict[str, Any]): Dictionary of accumulated tool calls.

​

_execute_tools_sync_with_status_accumulator

def _execute_tools_sync_with_status_accumulator(
    self,
    accumulated_tool_calls: Dict[str, Any],
    tool_call_records: List[ToolCallingRecord]
):

​

_execute_tool_from_stream_data

def _execute_tool_from_stream_data(self, tool_call_data: Dict[str, Any]):

​

_create_error_response

def _create_error_response(
    self,
    error_message: str,
    tool_call_records: List[ToolCallingRecord]
):

​

_record_assistant_tool_calls_message

def _record_assistant_tool_calls_message(self, accumulated_tool_calls: Dict[str, Any], content: str = ''):

​

_create_streaming_response_with_accumulator

def _create_streaming_response_with_accumulator(
    self,
    accumulator: StreamContentAccumulator,
    new_content: str,
    step_token_usage: Dict[str, int],
    response_id: str = '',
    tool_call_records: Optional[List[ToolCallingRecord]] = None
):

​

get_usage_dict

def get_usage_dict(self, output_messages: List[BaseMessage], prompt_tokens: int):

• output_messages (list): List of output messages.
• prompt_tokens (int): Number of input prompt tokens.

​

add_model_scheduling_strategy

def add_model_scheduling_strategy(self, name: str, strategy_fn: Callable):

• name (str): The name of the strategy.
• strategy_fn (Callable): The scheduling strategy function.

​

clone

def clone(self, with_memory: bool = False):

• with_memory (bool): Whether to copy the memory (conversation history) to the new agent. If True, the new agent will have the same conversation history. If False, the new agent will have a fresh memory with only the system message. (default: :obj:False)

​

_clone_tools

def _clone_tools(self):

• List of cloned tools/functions
• List of RegisteredAgentToolkit instances need registration

​

repr

def __repr__(self):

​

to_mcp

def to_mcp(
    self,
    name: str = 'CAMEL-ChatAgent',
    description: str = 'A helpful assistant using the CAMEL AI framework.',
    dependencies: Optional[List[str]] = None,
    host: str = 'localhost',
    port: int = 8000
):

• name (str): Name of the MCP server. (default: :obj:CAMEL-ChatAgent)
• description (Optional[List[str]]): Description of the agent. If None, a generic description is used. (default: :obj:A helpful assistant using the CAMEL AI framework.)
• dependencies (Optional[List[str]]): Additional dependencies for the MCP server. (default: :obj:None)
• host (str): Host to bind to for HTTP transport. (default: :obj:localhost)
• port (int): Port to bind to for HTTP transport. (default: :obj:8000)