autogen_ext.experimental.task_centric_memory.utils#

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

基类: object

一个将以任务为中心的内存与代理或团队结合的最小包装器。应用程序可以使用 Apprentice 类，也可以直接实例化并调用内存控制器，将此作为示例。

参数:

client – 用于调用模型的客户端。
config –
一个可选的字典，可用于覆盖以下值：
- name_of_agent_or_team: 用于分配任务的目标代理或团队的名称。
- disable_prefix_caching: 如果为 True，则通过在第一条消息前加上随机整数来禁用前缀缓存。
- MemoryController: 传递给 MemoryController 的配置字典。
logger – 一个可选的日志记录器。如果为 None，将创建一个默认的日志记录器。

reset_memory() → None[source]#: 重置内存库。

async handle_user_message(text: str, should_await: bool = True) → str[source]#: 处理用户消息，提取任何建议并为代理分配任务。

async add_task_solution_pair_to_memory(task: str, solution: str) → None[source]#: 向内存库添加一个任务-解决方案对，以便以后作为组合的洞察力一起检索。当洞察力是解决给定类型任务的演示时，这非常有用。

async assign_task(task: str, use_memory: bool = True, should_await: bool = True) → str[source]#: 向代理分配任务，以及任何相关的洞察/记忆。

async train_on_task(task: str, expected_answer: str) → None[source]#: 重复向完成代理分配任务，并通过创建有用的洞察作为记忆来尝试从失败中学习。

async assign_task_to_agent_or_team(task: str) → Tuple[str, str][source]#: 将给定任务传递给目标代理或团队。

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

基类：ChatCompletionClient

一个支持对调用 LLM 客户端的代码进行快速、大规模测试的聊天完成客户端。

支持两种模式

"record"（记录）：委托给底层客户端，同时记录输入消息和响应，并在调用 finalize() 时将其保存到磁盘。

"replay"（回放）：从磁盘加载先前记录的消息和响应，然后在每次调用时检查其消息是否与记录的消息匹配，并返回记录的响应。

记录的数据以记录的 JSON 列表形式存储。每个记录都是一个字典，包含一个“mode”字段（“create”或“create_stream”），一个序列化的消息列表，以及一个“response”（用于 create 调用）或一个“stream”（create_stream 调用的流式输出列表）。

ReplayChatCompletionClient 和 ChatCompletionCache 执行类似的操作，但有显著差异

ReplayChatCompletionClient 以指定的顺序回放预定义的响应，不记录任何内容或检查发送给客户端的消息。

ChatCompletionCache 缓存响应并为以前出现过的消息进行回放，无论顺序如何，并为任何未缓存的消息调用基本客户端。

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

从模型创建单个响应。

参数:

messages (Sequence[LLMMessage]) – 要发送给模型的消息。
tools (Sequence[Tool | ToolSchema], 可选) – 与模型一起使用的工具。默认为 []。
tool_choice (Tool | Literal["auto", "required", "none"], 可选) – 单个 Tool 对象，用于强制模型使用，"auto" 允许模型选择任何可用工具，"required" 强制使用工具，"none" 禁用工具使用。默认为 "auto"。
json_output (Optional[bool | type[BaseModel]], 可选) – 是否使用 JSON 模式、结构化输出或两者都不使用。默认为 None。如果设置为 Pydantic BaseModel 类型，它将用作结构化输出的输出类型。如果设置为布尔值，它将用于确定是否使用 JSON 模式。如果设置为 True，请确保在指令或提示中指示模型生成 JSON 输出。
extra_create_args (Mapping[str, Any], 可选) – 传递给底层客户端的额外参数。默认为 {}。
cancellation_token (Optional[CancellationToken], 可选) – 用于取消的令牌。默认为 None。

返回:

CreateResult – 模型调用的结果。

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

从模型创建字符串块流，以 CreateResult 结束。

参数:

messages (Sequence[LLMMessage]) – 要发送给模型的消息。
tools (Sequence[Tool | ToolSchema], 可选) – 与模型一起使用的工具。默认为 []。
tool_choice (Tool | Literal["auto", "required", "none"], 可选) – 单个 Tool 对象，用于强制模型使用，"auto" 允许模型选择任何可用工具，"required" 强制使用工具，"none" 禁用工具使用。默认为 "auto"。
json_output (Optional[bool | type[BaseModel]], 可选) –
是否使用 JSON 模式、结构化输出或两者都不使用。默认为 None。如果设置为 Pydantic BaseModel 类型，它将用作结构化输出的输出类型。如果设置为布尔值，它将用于确定是否使用 JSON 模式。如果设置为 True，请确保在指令或提示中指示模型生成 JSON 输出。
extra_create_args (Mapping[str, Any], 可选) – 传递给底层客户端的额外参数。默认为 {}。
cancellation_token (Optional[CancellationToken], 可选) – 用于取消的令牌。默认为 None。

返回:

AsyncGenerator[Union[str, CreateResult], None] – 生成器，产生字符串块并以 CreateResult 结束。

async close() → None[source]#

actual_usage() → RequestUsage[source]#

total_usage() → RequestUsage[source]#

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

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

property capabilities: ModelCapabilities#

property model_info: ModelInfo#

finalize() → None[source]#: 在记录模式下，将累积的记录保存到磁盘。在回放模式下，确保所有记录都已检查。

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

基类: object

运行基本测试，并确定任务成功与否，不受字符串匹配的限制。

参数:

client – 用于调用模型的客户端。
logger – 一个可选的日志记录器。如果为 None，则不执行日志记录。

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

async call_model(summary: str, user_content: str | List[str | Image], system_message_content: str | None = None, keep_these_messages: bool = True) → str[source]#: 使用给定输入调用模型客户端并返回响应。

async is_response_correct(task_description: str, response_to_be_graded: str, correct_answer: str) → Tuple[bool, str][source]#: 确定响应是否与任务的正确答案等效。

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

基类: object

将文本和图像记录到一组 HTML 页面中，每个函数/方法一个页面，并在调用树中相互链接。

参数:

config –

一个可选的字典，可用于覆盖以下值：

level: 日志记录级别，可以是 DEBUG、INFO、WARNING、ERROR、CRITICAL 或 NONE 之一。
path: 存储日志文件的目录路径。

finalize() → None[source]#

debug(line: str) → None[source]#: 如果调试级别 <= DEBUG，则向当前页面添加 DEBUG 文本。

info(line: str) → None[source]#: 如果调试级别 <= INFO，则向当前页面添加 INFO 文本。

warning(line: str) → None[source]#: 如果调试级别 <= WARNING，则向当前页面添加 WARNING 文本。

error(line: str) → None[source]#: 如果调试级别 <= ERROR，则向当前页面添加 ERROR 文本。

critical(line: str) → None[source]#: 如果调试级别 <= CRITICAL，则向当前页面添加 CRITICAL 文本。

log_message_content(message_content: str | List[str | Image] | List[FunctionCall] | List[FunctionExecutionResult], summary: str) → None[source]#: 添加一个包含消息内容（包括任何图像）的页面。

log_dict_list(content: List[Mapping[str, Any]], summary: str) → None[source]#: 添加一个包含字典列表的页面。

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]#: 将发送给模型的MESSAGES和TaskResult响应记录到新页面。

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]#: 将发送给模型的MESSAGES和TaskResult响应记录到新页面。

log_link_to_local_file(file_path: str) → str[source]#: 返回日志中本地文件的链接。

add_link_to_image(description: str, source_image_path: str) → None[source]#: 在页面中插入一个指向图像的缩略图链接。

flush(finished: bool = False) → None[source]#: 将日志的当前状态写入磁盘。

enter_function() → Page | None[source]#: 添加一个对应于当前函数调用的新页面。

leave_function() → None[source]#: 完成对应于当前函数调用的页面。

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

基类：Memory

赋予 AssistantAgent 快速从用户教学、提示和建议中学习的能力。

使用步骤

实例化 MemoryController。

实例化 Teachability，并将内存控制器作为参数传递。

实例化 AssistantAgent，将 teachability 实例（封装在列表中）作为内存参数传递。

像往常一样使用 AssistantAgent，例如与用户聊天。

property name: str#: 获取内存实例标识符。

async update_context(model_context: ChatCompletionContext) → UpdateContextResult[source]#: 从上次用户交互中提取任何建议并存储在内存中，并将任何相关记忆添加到模型上下文中。

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None[source]#: 尝试从传递的内容中提取任何建议并将其添加到内存中。

async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult[source]#: 返回与查询相关的任何记忆。

async clear() → None[source]#: 清除内存中的所有条目。

async close() → None[source]#: 清理内存资源。

class ApprenticeConfig[source]#

基类: TypedDict

name_of_agent_or_team: str#

disable_prefix_caching: bool#

MemoryController: MemoryControllerConfig#

class PageLoggerConfig[source]#

基类: TypedDict

level: str#

path: str#