autogen_ext.experimental.task_centric_memory.utils#

class Apprentice(client: ChatCompletionClient, config: ApprenticeConfig | None = None, logger: PageLogger | None = None)[source]#

Bases: object

A minimal wrapper combining task-centric memory with an agent or team. Applications may use the Apprentice class, or they may directly instantiate and call the Memory Controller using this class as an example.

Parameters:

client – The client to call the model.
config –
An optional dict that can be used to override the following values:
- name_of_agent_or_team: The name of the target agent or team for assigning tasks to.
- disable_prefix_caching: True to disable prefix caching by prepending random ints to the first message.
- MemoryController: A config dict passed to MemoryController.
logger – An optional logger. If None, a default logger will be created.

async add_task_solution_pair_to_memory(task: str, solution: str) → None[source]#: Adds a task-solution pair to the memory bank, to be retrieved together later as a combined insight. This is useful when the insight is a demonstration of how to solve a given type of task.

async assign_task(task: str, use_memory: bool = True, should_await: bool = True) → str[source]#: Assigns a task to the agent, along with any relevant insights/memories.

async assign_task_to_agent_or_team(task: str) → Tuple[str, str][source]#: Passes the given task to the target agent or team.

async handle_user_message(text: str, should_await: bool = True) → str[source]#: Handles a user message, extracting any advice and assigning a task to the agent.

reset_memory() → None[source]#: Resets the memory bank.

async train_on_task(task: str, expected_answer: str) → None[source]#: Repeatedly assigns a task to the completion agent, and tries to learn from failures by creating useful insights as memories.

class ChatCompletionClientRecorder(client: ChatCompletionClient, mode: Literal['record', 'replay'], session_file_path: str, logger: PageLogger | None = None)[source]#

Bases: ChatCompletionClient

A chat completion client that supports fast, large-scale tests of code calling LLM clients.

Two modes are supported:

“record”: delegates to the underlying client while also recording the input messages and responses, which are saved to disk when finalize() is called.

“replay”: loads previously recorded message and responses from disk, then on each call checks that its message matches the recorded message, and returns the recorded response.

The recorded data is stored as a JSON list of records. Each record is a dictionary with a “mode” field (either “create” or “create_stream”), a serialized list of messages, and either a “response” (for create calls) or a “stream” (a list of streamed outputs for create_stream calls).

ReplayChatCompletionClient and ChatCompletionCache do similar things, but with significant differences: - ReplayChatCompletionClient replays pre-defined responses in a specified order without recording anything or checking the messages sent to the client. - ChatCompletionCache caches responses and replays them for messages that have been seen before, regardless of order, and calls the base client for any uncached messages.

actual_usage() → RequestUsage[source]#

property capabilities: ModelCapabilities#

async close() → None[source]#

count_tokens(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = []) → int[source]#

async create(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = [], json_output: bool | None = None, extra_create_args: Mapping[str, Any] = {}, cancellation_token: CancellationToken | None = None) → CreateResult[source]#

create_stream(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = [], json_output: bool | None = None, extra_create_args: Mapping[str, Any] = {}, cancellation_token: CancellationToken | None = None) → AsyncGenerator[str | CreateResult, None][source]#

finalize() → None[source]#: In record mode, saves the accumulated records to disk. In replay mode, makes sure all the records were checked.

property model_info: ModelInfo#

remaining_tokens(messages: Sequence[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[Tool | ToolSchema] = []) → int[source]#

total_usage() → RequestUsage[source]#

class Grader(client: ChatCompletionClient, logger: PageLogger | None = None)[source]#

Bases: object

Runs basic tests, and determines task success without limitation to string matches.

Parameters:

client – The client to call the model.
logger – An optional logger. If None, no logging will be performed.

async call_model(summary: str, user_content: str | List[str | Image], system_message_content: str | None = None, keep_these_messages: bool = True) → str[source]#: Calls the model client with the given input and returns the response.

async is_response_correct(task_description: str, response_to_be_graded: str, correct_answer: str) → Tuple[bool, str][source]#: Determines whether the response is equivalent to the task’s correct answer.

async test_apprentice(apprentice: Apprentice, task_description: str, expected_answer: str, num_trials: int, use_memory: bool, client: ChatCompletionClient) → Tuple[int, int][source]#

class PageLogger(config: PageLoggerConfig | None = None)[source]#

Bases: object

Logs text and images to a set of HTML pages, one per function/method, linked to each other in a call tree.

Parameters:

config –

An optional dict that can be used to override the following values:

level: The logging level, one of DEBUG, INFO, WARNING, ERROR, CRITICAL, or NONE.
path: The path to the directory where the log files will be written.

add_link_to_image(description: str, source_image_path: str) → None[source]#: Inserts a thumbnail link to an image to the page.

critical(line: str) → None[source]#: Adds CRITICAL text to the current page if debugging level <= CRITICAL.

debug(line: str) → None[source]#: Adds DEBUG text to the current page if debugging level <= DEBUG.

enter_function() → Page | None[source]#: Adds a new page corresponding to the current function call.

error(line: str) → None[source]#: Adds ERROR text to the current page if debugging level <= ERROR.

finalize() → None[source]#

flush(finished: bool = False) → None[source]#: Writes the current state of the log to disk.

info(line: str) → None[source]#: Adds INFO text to the current page if debugging level <= INFO.

leave_function() → None[source]#: Finishes the page corresponding to the current function call.

log_dict_list(content: List[Mapping[str, Any]], summary: str) → None[source]#: Adds a page containing a list of dicts.

log_link_to_local_file(file_path: str) → str[source]#: Returns a link to a local file in the log.

log_message_content(message_content: str | List[str | Image] | List[FunctionCall] | List[FunctionExecutionResult], summary: str) → None[source]#: Adds a page containing the message’s content, including any images.

log_model_call(summary: str, input_messages: List[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], response: CreateResult) → Page | None[source]#: Logs messages sent to a model and the TaskResult response to a new page.

log_model_task(summary: str, input_messages: List[Annotated[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], task_result: TaskResult) → Page | None[source]#: Logs messages sent to a model and the TaskResult response to a new page.

warning(line: str) → None[source]#: Adds WARNING text to the current page if debugging level <= WARNING.

class Teachability(memory_controller: MemoryController, name: str | None = None)[source]#

Bases: Memory

Gives an AssistantAgent the ability to learn quickly from user teachings, hints, and advice.

Steps for usage: 1. Instantiate MemoryController. 2. Instantiate Teachability, passing the memory controller as a parameter. 3. Instantiate an AssistantAgent, passing the teachability instance (wrapped in a list) as the memory parameter. 4. Use the AssistantAgent as usual, such as for chatting with the user.

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None[source]#: Tries to extract any advice from the passed content and add it to memory.

async clear() → None[source]#: Clear all entries from memory.

async close() → None[source]#: Clean up memory resources.

property name: str#: Get the memory instance identifier.

async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult[source]#: Returns any memories that seem relevant to the query.

async update_context(model_context: ChatCompletionContext) → UpdateContextResult[source]#: Extracts any advice from the last user turn to be stored in memory, and adds any relevant memories to the model context.