autogen.agents.experimental.TelegramAgent

​

TelegramAgent

TelegramAgent(
    name: str,
    system_message: str | None = None,
    *,
    api_id: str,
    api_hash: str,
    chat_id: str,
    has_writing_instructions: bool = True,
    **kwargs: Any
)

An agent that can send messages and retrieve messages on Telegram.
Initialize the TelegramAgent.

​

Class Attributes

​

DEFAULT_SYSTEM_MESSAGE

​

Instance Attributes

​

chat_messages

A dictionary of conversations from agent to list of messages.

​

code_executor

The code executor used by this agent. Returns None if code execution is disabled.

​

description

Get the description of the agent.

​

function_map

Return the function map.

​

name

Get the name of the agent.

​

system_message

Return the system message.

​

tools

Get the agent’s tools (registered for LLM)

Note this is a copy of the tools list, use add_tool and remove_tool to modify the tools list.

​

use_docker

Bool value of whether to use docker to execute the code, or str value of the docker image name to use, or None when code execution is disabled.

​

Instance Methods

​

a_check_termination_and_human_reply

a_check_termination_and_human_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, str | None]

(async) Check if the conversation should be terminated, and if human reply is provided.
This method checks for conditions that require the conversation to be terminated, such as reaching a maximum number of consecutive auto-replies or encountering a termination message. Additionally, it prompts for and processes human input based on the configured human input mode, which can be ‘ALWAYS’, ‘NEVER’, or ‘TERMINATE’. The method also manages the consecutive auto-reply counter for the conversation and prints relevant messages based on the human input received.

​

a_execute_function

a_execute_function(
    self,
    func_call: dict[str, Any],
    call_id: str | None = None,
    verbose: bool = False
) -> tuple[bool, dict[str, Any]]

Execute an async function call and return the result.
Override this function to modify the way async functions and tools are executed.

​

a_generate_function_call_reply

a_generate_function_call_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, dict[str, Any] | None]

Generate a reply using async function call.
”function_call” replaced by “tool_calls” as of OpenAI API v1.1.0 See https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions

​

a_generate_init_message

a_generate_init_message(
    self,
    message: str | dict[str, Any] | None,
    **kwargs: Any
) -> dict[str, Any] | str

Generate the initial message for the agent.
If message is None, input() will be called to get the initial message.

​

a_generate_oai_reply

a_generate_oai_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, str | dict[str, Any] | None]

Generate a reply using autogen.oai asynchronously.

​

a_generate_reply

a_generate_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: ForwardRef('Agent') | None = None,
    **kwargs: Any
) -> str | dict[str, Any] | None

(async) Reply based on the conversation history and the sender.
Either messages or sender must be provided.
Register a reply_func with None as one trigger for it to be activated when messages is non-empty and sender is None.
Use registered auto reply functions to generate replies.
By default, the following functions are checked in order:
1. check_termination_and_human_reply 2. generate_function_call_reply 3. generate_tool_calls_reply 4. generate_code_execution_reply 5. generate_oai_reply Every function returns a tuple (final, reply).
When a function returns final=False, the next function will be checked.
So by default, termination and human reply will be checked first.
If not terminating and human reply is skipped, execute function or code and return the result.
AI replies are generated only when no code execution is performed.

​

a_generate_tool_calls_reply

a_generate_tool_calls_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, dict[str, Any] | None]

Generate a reply using async function call.

​

a_get_human_input

a_get_human_input(self, prompt: str) -> str

(Async) Get human input.
Override this method to customize the way to get human input.

​

a_initiate_chat

a_initiate_chat(
    self,
    recipient: ConversableAgent,
    clear_history: bool = True,
    silent: bool | None = False,
    cache: AbstractCache | None = None,
    max_turns: int | None = None,
    summary_method: str | Callable[..., Any] | None = 'last_msg',
    summary_args: dict[str, Any] | None = {},
    message: str | Callable[..., Any] | None = None,
    **kwargs: Any
) -> ChatResult

(async) Initiate a chat with the recipient agent.
Reset the consecutive auto reply counter.
If clear_history is True, the chat history with the recipient agent will be cleared.
a_generate_init_message is called to generate the initial message for the agent.

​

a_receive

a_receive(
    self,
    message: dict[str, Any] | str,
    sender: Agent,
    request_reply: bool | None = None,
    silent: bool | None = False
) -> 

(async) Receive a message from another agent.
Once a message is received, this function sends a reply to the sender or stop.
The reply can be generated automatically or entered manually by a human.

​

a_run

a_run(
    self,
    message: str,
    *,
    tools: Tool | Iterable[Tool] | None = None,
    executor_kwargs: dict[str, Any] | None = None,
    max_turns: int | None = None,
    msg_to: Literal['agent', 'user'] = 'agent',
    clear_history: bool = False,
    user_input: bool = True,
    summary_method: str | Callable[..., Any] | None = 'last_msg'
) -> ChatResult

Run a chat asynchronously with the agent using the given message.
A second agent will be created to represent the user, this agent will by known by the name ‘user’.
The user can terminate the conversation when prompted or, if agent’s reply contains ‘TERMINATE’, it will terminate.

​

a_send

a_send(
    self,
    message: dict[str, Any] | str,
    recipient: Agent,
    request_reply: bool | None = None,
    silent: bool | None = False
) -> 

(async) Send a message to another agent.

​

can_execute_function

can_execute_function(self, name: list[str] | str) -> bool

Whether the agent can execute the function.

​

chat_messages_for_summary

chat_messages_for_summary(self, agent: Agent) -> list[dict[str, Any]]

A list of messages as a conversation to summarize.

​

check_termination_and_human_reply

check_termination_and_human_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, str | None]

Check if the conversation should be terminated, and if human reply is provided.
This method checks for conditions that require the conversation to be terminated, such as reaching a maximum number of consecutive auto-replies or encountering a termination message. Additionally, it prompts for and processes human input based on the configured human input mode, which can be ‘ALWAYS’, ‘NEVER’, or ‘TERMINATE’. The method also manages the consecutive auto-reply counter for the conversation and prints relevant messages based on the human input received.

​

clear_history

clear_history(
    self,
    recipient: Agent | None = None,
    nr_messages_to_preserve: int | None = None
) -> 

Clear the chat history of the agent.

​

execute_code_blocks

execute_code_blocks(self, code_blocks) -> 

Execute the code blocks and return the result.

​

execute_function

execute_function(
    self,
    func_call: dict[str, Any],
    call_id: str | None = None,
    verbose: bool = False
) -> tuple[bool, dict[str, Any]]

Execute a function call and return the result.
Override this function to modify the way to execute function and tool calls.

​

generate_code_execution_reply

generate_code_execution_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: dict[str, Any] | Literal[False] | None = None
) -> 

Generate a reply using code execution.

​

generate_function_call_reply

generate_function_call_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, dict[str, Any] | None]

Generate a reply using function call.
”function_call” replaced by “tool_calls” as of OpenAI API v1.1.0 See https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions

​

generate_init_message

generate_init_message(
    self,
    message: str | dict[str, Any] | None,
    **kwargs: Any
) -> dict[str, Any] | str

Generate the initial message for the agent.
If message is None, input() will be called to get the initial message.

​

generate_oai_reply

generate_oai_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: OpenAIWrapper | None = None
) -> tuple[bool, str | dict[str, Any] | None]

Generate a reply using autogen.oai.

​

generate_reply

generate_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: ForwardRef('Agent') | None = None,
    **kwargs: Any
) -> str | dict[str, Any] | None

Reply based on the conversation history and the sender.
Either messages or sender must be provided.
Register a reply_func with None as one trigger for it to be activated when messages is non-empty and sender is None.
Use registered auto reply functions to generate replies.
By default, the following functions are checked in order:
1. check_termination_and_human_reply 2. generate_function_call_reply (deprecated in favor of tool_calls) 3. generate_tool_calls_reply 4. generate_code_execution_reply 5. generate_oai_reply Every function returns a tuple (final, reply).
When a function returns final=False, the next function will be checked.
So by default, termination and human reply will be checked first.
If not terminating and human reply is skipped, execute function or code and return the result.
AI replies are generated only when no code execution is performed.

​

generate_tool_calls_reply

generate_tool_calls_reply(
    self,
    messages: list[dict[str, Any]] | None = None,
    sender: Agent | None = None,
    config: Any | None = None
) -> tuple[bool, dict[str, Any] | None]

Generate a reply using tool call.

​

get_actual_usage

get_actual_usage(self) -> dict[str, int] | None

Get the actual usage summary.

​

get_chat_results

get_chat_results(self, chat_index: int | None = None) -> list[ChatResult] | ChatResult

A summary from the finished chats of particular agents.

​

get_context

get_context(
    self,
    key: str,
    default: Any = None
) -> Any

Get a context variable by key.

​

get_human_input

get_human_input(self, prompt: str) -> str

Get human input.
Override this method to customize the way to get human input.

​

get_total_usage

get_total_usage(self) -> dict[str, int] | None

Get the total usage summary.

​

initiate_chat

initiate_chat(
    self,
    recipient: ConversableAgent,
    clear_history: bool = True,
    silent: bool | None = False,
    cache: AbstractCache | None = None,
    max_turns: int | None = None,
    summary_method: str | Callable[..., Any] | None = 'last_msg',
    summary_args: dict[str, Any] | None = {},
    message: dict[str, Any] | str | Callable[..., Any] | None = None,
    **kwargs: Any
) -> ChatResult

Initiate a chat with the recipient agent.
Reset the consecutive auto reply counter.
If clear_history is True, the chat history with the recipient agent will be cleared.

​

initiate_chats

initiate_chats(self, chat_queue: list[dict[str, Any]]) -> list[ChatResult]

(Experimental) Initiate chats with multiple agents.

​

last_message

last_message(self, agent: Agent | None = None) -> dict[str, Any] | None

The last message exchanged with the agent.

​

max_consecutive_auto_reply

max_consecutive_auto_reply(self, sender: Agent | None = None) -> int

The maximum number of consecutive auto replies.

​

pop_context

pop_context(
    self,
    key: str,
    default: Any = None
) -> Any

Remove and return a context variable.

​

print_usage_summary

print_usage_summary(self, mode: list[str] | str = ['actual', 'total']) -> None

Print the usage summary.

​

process_all_messages_before_reply

process_all_messages_before_reply(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]

Calls any registered capability hooks to process all messages, potentially modifying the messages.

​

process_last_received_message

process_last_received_message(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]

Calls any registered capability hooks to use and potentially modify the text of the last message, as long as the last message is not a function call or exit command.

​

receive

receive(
    self,
    message: dict[str, Any] | str,
    sender: Agent,
    request_reply: bool | None = None,
    silent: bool | None = False
) -> 

Receive a message from another agent.
Once a message is received, this function sends a reply to the sender or stop.
The reply can be generated automatically or entered manually by a human.

​

register_for_execution

register_for_execution(
    self,
    name: str | None = None,
    description: str | None = None,
    *,
    serialize: bool = True,
    silent_override: bool = False
) -> Callable[[~F | Tool], Tool]

Decorator factory for registering a function to be executed by an agent.
It’s return value is used to decorate a function to be registered to the agent.

​

register_for_llm

register_for_llm(
    self,
    *,
    name: str | None = None,
    description: str | None = None,
    api_style: Literal['function', 'tool'] = 'tool',
    silent_override: bool = False
) -> Callable[[~F | Tool], Tool]

Decorator factory for registering a function to be used by an agent.
It’s return value is used to decorate a function to be registered to the agent. The function uses type hints to specify the arguments and return type. The function name is used as the default name for the function, but a custom name can be provided. The function description is used to describe the function in the agent’s configuration.

​

register_function

register_function(
    self,
    function_map: dict[str, Callable[..., Any]],
    silent_override: bool = False
) -> 

Register functions to the agent.

​

register_hook

register_hook(
    self,
    hookable_method: str,
    hook: Callable
) -> 

Registers a hook to be called by a hookable method, in order to add a capability to the agent.
Registered hooks are kept in lists (one per hookable method), and are called in their order of registration.

​

register_model_client

register_model_client(
    self,
    model_client_cls: ModelClient,
    **kwargs: Any
) -> 

Register a model client.

​

register_nested_chats

register_nested_chats(
    self,
    chat_queue: list[dict[str, Any]],
    trigger: type[Agent] | str | Agent | Callable[[Agent], bool] | list,
    reply_func_from_nested_chats: str | Callable[..., Any] = 'summary_from_nested_chats',
    position: int = 2,
    use_async: bool | None = None,
    **kwargs: Any
) -> None

Register a nested chat reply function.

​

register_reply

register_reply(
    self,
    trigger: type[Agent] | str | Agent | Callable[[Agent], bool] | list,
    reply_func: Callable,
    position: int = 0,
    config: Any | None = None,
    reset_config: Callable[..., Any] | None = None,
    *,
    ignore_async_in_sync_chat: bool = False,
    remove_other_reply_funcs: bool = False
) -> 

Register a reply function.
The reply function will be called when the trigger matches the sender.
The function registered later will be checked earlier by default.
To change the order, set the position to a positive integer.
Both sync and async reply functions can be registered. The sync reply function will be triggered from both sync and async chats. However, an async reply function will only be triggered from async chats (initiated with ConversableAgent.a_initiate_chat). If an async reply function is registered and a chat is initialized with a sync function, ignore_async_in_sync_chat determines the behaviour as follows:
if ignore_async_in_sync_chat is set to False (default value), an exception will be raised, and if ignore_async_in_sync_chat is set to True, the reply function will be ignored.

​

remove_tool_for_llm

remove_tool_for_llm(self, tool: Tool) -> None

Remove a tool (register for LLM tool)

​

replace_reply_func

replace_reply_func(
    self,
    old_reply_func: Callable,
    new_reply_func: Callable
) -> 

Replace a registered reply function with a new one.

​

reset

reset(self) -> None

Reset the agent.

​

reset_consecutive_auto_reply_counter

reset_consecutive_auto_reply_counter(self, sender: Agent | None = None) -> 

Reset the consecutive_auto_reply_counter of the sender.

​

run

run(
    self,
    message: str,
    *,
    tools: Tool | Iterable[Tool] | None = None,
    executor_kwargs: dict[str, Any] | None = None,
    max_turns: int | None = None,
    msg_to: Literal['agent', 'user'] = 'agent',
    clear_history: bool = False,
    user_input: bool = True,
    summary_method: str | Callable[..., Any] | None = 'last_msg'
) -> ChatResult

Run a chat with the agent using the given message.
A second agent will be created to represent the user, this agent will by known by the name ‘user’. This agent does not have code execution enabled by default, if needed pass the code execution config in with the executor_kwargs parameter.
The user can terminate the conversation when prompted or, if agent’s reply contains ‘TERMINATE’, it will terminate.

​

run_code

run_code(
    self,
    code: str,
    **kwargs: Any
) -> tuple[int, str, str | None]

Run the code and return the result.
Override this function to modify the way to run the code.

​

send

send(
    self,
    message: dict[str, Any] | str,
    recipient: Agent,
    request_reply: bool | None = None,
    silent: bool | None = False
) -> 

Send a message to another agent.

​

set_context

set_context(
    self,
    key: str,
    value: Any
) -> None

Set a context variable.

​

stop_reply_at_receive

stop_reply_at_receive(self, sender: Agent | None = None) -> 

Reset the reply_at_receive of the sender.

​

update_agent_state_before_reply

update_agent_state_before_reply(self, messages: list[dict[str, Any]]) -> None

Calls any registered capability hooks to update the agent’s state.
Primarily used to update context variables.
Will, potentially, modify the messages.

​

update_context

update_context(self, context_variables: dict[str, Any]) -> None

Update multiple context variables at once.

​

update_function_signature

update_function_signature(
    self,
    func_sig: dict[str, Any] | str,
    is_remove: None,
    silent_override: bool = False
) -> 

Update a function_signature in the LLM configuration for function_call.

​

update_max_consecutive_auto_reply

update_max_consecutive_auto_reply(
    self,
    value: int,
    sender: Agent | None = None
) -> 

Update the maximum number of consecutive auto replies.

​

update_system_message

update_system_message(self, system_message: str) -> None

Update the system message.

​

update_tool_signature

update_tool_signature(
    self,
    tool_sig: dict[str, Any] | str,
    is_remove: bool,
    silent_override: bool = False
) -> 

Update a tool_signature in the LLM configuration for tool_call.