autogen_ext.experimental.task_centric_memory.utils#

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

基类: object

一个最小化的包装器，将以任务为中心的内存与代理或团队结合在一起。应用程序可以使用 Apprentice 类，或者他们可以以此类为例直接实例化和调用 Memory Controller。

参数:

client – 调用模型的客户端。
config –
一个可选的 dict，可用于覆盖以下值
- name_of_agent_or_team: 用于分配任务的目标代理或团队的名称。
- disable_prefix_caching: True 表示禁用前缀缓存，方法是在第一条消息前添加随机整数。
- MemoryController: 传递给 MemoryController 的 config dict。
logger – 一个可选的 logger。如果为 None，将创建一个默认的 logger。

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 assign_task_to_agent_or_team(task: str) → Tuple[str, str][source]#: 将给定的任务传递给目标代理或团队。

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

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

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

class ApprenticeConfig[source]#

基类： TypedDict

MemoryController: MemoryControllerConfig#

disable_prefix_caching: bool#

name_of_agent_or_team: str#

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 缓存响应并为之前见过的消息重放它们，无论顺序如何，并为任何未缓存的消息调用基本客户端。

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 | type[BaseModel] | None = None, extra_create_args: Mapping[str, Any] = {}, cancellation_token: CancellationToken | None = None) → CreateResult[source]#

从模型创建一个单独的响应。

参数:

messages (Sequence[LLMMessage]) – 要发送给模型的消息。
tools (Sequence[Tool | ToolSchema], optional) – 要与模型一起使用的工具。默认为 []。
json_output (Optional[bool | type[BaseModel]], optional) – 是否使用 JSON 模式、结构化输出，或者都不使用。默认为 None。如果设置为 Pydantic BaseModel 类型，它将用作结构化输出的输出类型。如果设置为布尔值，它将用于确定是否使用 JSON 模式。如果设置为 True，请确保指示模型在指令或提示中生成 JSON 输出。
extra_create_args (Mapping[str, Any], optional) – 传递给底层客户端的额外参数。默认为 {}。
cancellation_token (Optional[CancellationToken], optional) – 用于取消的令牌。默认为 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) → AsyncGenerator[str | CreateResult, None][source]#

创建一个来自模型的字符串块流，以 CreateResult 结尾。

参数:

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

返回值:

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

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

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]#

基类: object

运行基本测试，并确定任务是否成功，不限于字符串匹配。

参数:

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

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]#: 确定响应是否等同于任务的正确答案。

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]#

基类: object

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

参数:

config –

一个可选的 dict，可用于覆盖以下值

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

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

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

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

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

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

finalize() → None[source]#

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

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

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

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

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

log_message_content(message_content: str | List[str | Image] | List[FunctionCall] | List[FunctionExecutionResult], 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]#: 将发送到模型的消息和 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]#: 将发送到模型的消息和 TaskResult 响应记录到一个新页面。

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

class PageLoggerConfig[source]#

基类： TypedDict

level: str#

path: str#

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

基类: Memory

使 AssistantAgent 能够从用户的教导、提示和建议中快速学习。

使用步骤

实例化 MemoryController。

实例化 Teachability，并将 memory controller 作为参数传递。

实例化一个 AssistantAgent，并将 teachability 实例（包装在列表中）作为 memory 参数传递。

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

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

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

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

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

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

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