autogen_ext.experimental.task_centric_memory.utils#

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

Basiert auf: object

Ein minimaler Wrapper, der aufgabenorientierten Speicher mit einem Agenten oder Team kombiniert. Anwendungen können die Apprentice-Klasse verwenden oder den Memory Controller direkt instanziieren und aufrufen, wobei diese Klasse als Beispiel dient.

Parameter:

client – Der Client zum Aufrufen des Modells.
config –
Ein optionales Dict, mit dem die folgenden Werte überschrieben werden können
- name_of_agent_or_team: Der Name des Zielagenten oder -teams, dem Aufgaben zugewiesen werden sollen.
- disable_prefix_caching: True, um das Caching von Präfixen zu deaktivieren, indem zufällige Ganzzahlen vor die erste Nachricht gestellt werden.
- MemoryController: Ein Dict mit Konfiguration, das an MemoryController übergeben wird.
logger – Ein optionaler Logger. Wenn None, wird ein Standardlogger erstellt.

reset_memory() → None[Quelle]#: Setzt die Speicherbank zurück.

async handle_user_message(text: str, should_await: bool = True) → str[Quelle]#: Behandelt eine Benutzernachricht, extrahiert Ratschläge und weist dem Agenten eine Aufgabe zu.

async add_task_solution_pair_to_memory(task: str, solution: str) → None[Quelle]#: Fügt ein Aufgaben-Lösungs-Paar zum Speicher hinzu, das später als kombinierte Erkenntnis abgerufen werden kann. Dies ist nützlich, wenn die Erkenntnis eine Demonstration ist, wie eine bestimmte Art von Aufgabe gelöst wird.

async assign_task(task: str, use_memory: bool = True, should_await: bool = True) → str[Quelle]#: Weist dem Agenten eine Aufgabe zu, zusammen mit relevanten Erkenntnissen/Erinnerungen.

async train_on_task(task: str, expected_answer: str) → None[Quelle]#: Weist dem Completion-Agenten wiederholt eine Aufgabe zu und versucht, aus Fehlern zu lernen, indem nützliche Erkenntnisse als Erinnerungen erstellt werden.

async assign_task_to_agent_or_team(task: str) → Tuple[str, str][Quelle]#: Übergibt die gegebene Aufgabe an den Zielagenten oder das Zielteam.

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

Bases: ChatCompletionClient

Ein Chat-Completion-Client, der schnelle, groß angelegte Tests von Code unterstützt, der LLM-Clients aufruft.

Zwei Modi werden unterstützt

„record“: delegiert an den zugrunde liegenden Client und zeichnet gleichzeitig die Eingabenachrichten und Antworten auf, die beim Aufruf von finalize() auf der Festplatte gespeichert werden.

„replay“: lädt zuvor aufgenommene Nachrichten und Antworten von der Festplatte, prüft dann bei jedem Aufruf, ob seine Nachricht mit der aufgezeichneten Nachricht übereinstimmt, und gibt die aufgezeichnete Antwort zurück.

Die aufgezeichneten Daten werden als JSON-Liste von Datensätzen gespeichert. Jeder Datensatz ist ein Dictionary mit einem Feld „mode“ (entweder „create“ oder „create_stream“), einer serialisierten Liste von Nachrichten und entweder einer „response“ (für create-Aufrufe) oder einem „stream“ (einer Liste von gestreamten Ausgaben für create_stream-Aufrufe).

ReplayChatCompletionClient und ChatCompletionCache tun ähnliche Dinge, aber mit signifikanten Unterschieden

ReplayChatCompletionClient spielt vordefinierte Antworten in einer bestimmten Reihenfolge ab, ohne etwas aufzuzeichnen oder die an den Client gesendeten Nachrichten zu überprüfen.

ChatCompletionCache speichert Antworten zwischen und spielt sie für zuvor gesehene Nachrichten ab, unabhängig von der Reihenfolge, und ruft den Basisclient für nicht zwischengespeicherte Nachrichten auf.

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

Creates a single response from the model.

Parameter:

messages (Sequence[LLMMessage]) – The messages to send to the model.
tools (Sequence[Tool | ToolSchema], optional) – The tools to use with the model. Defaults to [].
tool_choice (Tool | Literal["auto", "required", "none"], optional) – A single Tool object to force the model to use, “auto” to let the model choose any available tool, “required” to force tool usage, or “none” to disable tool usage. Defaults to “auto”.
json_output (Optional[bool | type[BaseModel]], optional) – Whether to use JSON mode, structured output, or neither. Defaults to None. If set to a Pydantic BaseModel type, it will be used as the output type for structured output. If set to a boolean, it will be used to determine whether to use JSON mode or not. If set to True, make sure to instruct the model to produce JSON output in the instruction or prompt.
extra_create_args (Mapping[str, Any], optional) – Extra arguments to pass to the underlying client. Defaults to {}.
cancellation_token (Optional[CancellationToken], optional) – A token for cancellation. Defaults to None.

Gibt zurück:

CreateResult – The result of the model call.

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

Creates a stream of string chunks from the model ending with a CreateResult.

Parameter:

messages (Sequence[LLMMessage]) – The messages to send to the model.
tools (Sequence[Tool | ToolSchema], optional) – The tools to use with the model. Defaults to [].
tool_choice (Tool | Literal["auto", "required", "none"], optional) – A single Tool object to force the model to use, “auto” to let the model choose any available tool, “required” to force tool usage, or “none” to disable tool usage. Defaults to “auto”.
json_output (Optional[bool | type[BaseModel]], optional) –
Whether to use JSON mode, structured output, or neither. Defaults to None. If set to a Pydantic BaseModel type, it will be used as the output type for structured output. If set to a boolean, it will be used to determine whether to use JSON mode or not. If set to True, make sure to instruct the model to produce JSON output in the instruction or prompt.
extra_create_args (Mapping[str, Any], optional) – Extra arguments to pass to the underlying client. Defaults to {}.
cancellation_token (Optional[CancellationToken], optional) – A token for cancellation. Defaults to None.

Gibt zurück:

AsyncGenerator[Union[str, CreateResult], None] – A generator that yields string chunks and ends with a CreateResult.

async close() → None[Quelle]#

actual_usage() → RequestUsage[Quelle]#

total_usage() → RequestUsage[Quelle]#

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

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

property capabilities: ModelCapabilities#

property model_info: ModelInfo#

finalize() → None[source]#: Speichert die angesammelten Aufzeichnungen im Aufnahmemodus auf der Festplatte. Stellt im Wiedergabemodus sicher, dass alle Aufzeichnungen überprüft wurden.

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

Basiert auf: object

Führt grundlegende Tests durch und bestimmt den Erfolg einer Aufgabe, ohne auf Zeichenfolgenübereinstimmungen beschränkt zu sein.

Parameter:

client – Der Client zum Aufrufen des Modells.
logger – Ein optionaler Logger. Wenn None, wird keine Protokollierung durchgeführt.

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]#: Ruft den Modellclient mit der angegebenen Eingabe auf und gibt die Antwort zurück.

async is_response_correct(task_description: str, response_to_be_graded: str, correct_answer: str) → Tuple[bool, str][source]#: Bestimmt, ob die Antwort der korrekten Antwort der Aufgabe entspricht.

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

Basiert auf: object

Protokolliert Text und Bilder in einer Reihe von HTML-Seiten, eine pro Funktion/Methode, die in einem Aufrufbaum miteinander verknüpft sind.

Parameter:

config –

Ein optionales Dict, mit dem die folgenden Werte überschrieben werden können

level: Die Protokollierungsstufe, eine von DEBUG, INFO, WARNING, ERROR, CRITICAL oder NONE.
path: Der Pfad zum Verzeichnis, in das die Protokolldateien geschrieben werden sollen.

finalize() → None[source]#

debug(line: str) → None[source]#: Fügt der aktuellen Seite DEBUG-Text hinzu, wenn die Debug-Stufe <= DEBUG ist.

info(line: str) → None[source]#: Fügt der aktuellen Seite INFO-Text hinzu, wenn die Debug-Stufe <= INFO ist.

warning(line: str) → None[source]#: Fügt der aktuellen Seite WARNING-Text hinzu, wenn die Debug-Stufe <= WARNING ist.

error(line: str) → None[source]#: Fügt der aktuellen Seite ERROR-Text hinzu, wenn die Debug-Stufe <= ERROR ist.

critical(line: str) → None[source]#: Fügt der aktuellen Seite CRITICAL-Text hinzu, wenn die Debug-Stufe <= CRITICAL ist.

log_message_content(message_content: str | List[str | Image] | List[FunctionCall] | List[FunctionExecutionResult], summary: str) → None[source]#: Fügt eine Seite mit dem Inhalt der Nachricht hinzu, einschließlich aller Bilder.

log_dict_list(content: List[Mapping[str, Any]], summary: str) → None[source]#: Fügt eine Seite mit einer Liste von Dictionaries hinzu.

log_model_call(summary: str, input_messages: List[Annotiert[SystemMessage | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], response: CreateResult) → Page | None[source]#: Protokolliert die an ein Modell gesendeten Nachrichten und die TaskResult-Antwort auf einer neuen Seite.

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]#: Protokolliert die an ein Modell gesendeten Nachrichten und die TaskResult-Antwort auf einer neuen Seite.

log_link_to_local_file(file_path: str) → str[source]#: Gibt einen Link zu einer lokalen Datei in der Protokollierung zurück.

add_link_to_image(description: str, source_image_path: str) → None[source]#: Fügt einen Miniaturbildlink zu einem Bild zur Seite ein.

flush(finished: bool = False) → None[source]#: Schreibt den aktuellen Zustand des Protokolls auf die Festplatte.

enter_function() → Page | None[source]#: Fügt eine neue Seite hinzu, die dem aktuellen Funktionsaufruf entspricht.

leave_function() → None[source]#: Schließt die Seite, die dem aktuellen Funktionsaufruf entspricht.

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

Bases: Memory

Verleiht einem AssistantAgent die Fähigkeit, schnell von Benutzereingaben, Hinweisen und Ratschlägen zu lernen.

Schritte zur Verwendung

Instanziieren Sie MemoryController.

Instanziieren Sie Teachability und übergeben Sie den MemoryController als Parameter.

Instanziieren Sie einen AssistantAgent und übergeben Sie die Teachability-Instanz (eingeschlossen in einer Liste) als Speicherparameter.

Verwenden Sie den AssistantAgent wie gewohnt, z. B. für Chats mit dem Benutzer.

property name: str#: Ruft den Speicherinstanz-Identifikator ab.

async update_context(model_context: ChatCompletionContext) → UpdateContextResult[source]#: Extrahiert Ratschläge aus der letzten Benutzereingabe, um sie im Speicher abzulegen, und fügt relevante Speicherinhalte in den Modellkontext ein.

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None[source]#: Versucht, Ratschläge aus dem übergebenen Inhalt zu extrahieren und im Speicher abzulegen.

async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult[source]#: Gibt Speicherinhalte zurück, die für die Abfrage relevant zu sein scheinen.

async clear() → None[source]#: Löschen Sie alle Einträge aus dem Speicher.

async close() → None[source]#: Speicherressourcen bereinigen.

class ApprenticeConfig[source]#

Bases: TypedDict

name_of_agent_or_team: str#

disable_prefix_caching: bool#

MemoryController: MemoryControllerConfig#

class PageLoggerConfig[source]#

Bases: TypedDict

level: str#

path: str#