autogen_ext.memory.canvas#
- class TextCanvas[Quelle]#
Bases:
BaseCanvasEin Canvas im Arbeitsspeicher, der Textdateien mit vollständiger Revisionshistorie speichert.
Warnung
Dies ist eine experimentelle API und kann sich in Zukunft ändern.
Neben den ursprünglichen CRUD-ähnlichen Operationen fügt diese erweiterte Implementierung hinzu:
apply_patch – wendet Patches mithilfe der
unidiffBibliothek für eine genaue Anwendung von Hunks und die Validierung von Kontextzeilen an.get_revision_content – zufälliger Zugriff auf jede historische Revision.
get_revision_diffs – erhält die Liste der Diffs, die zwischen jedem aufeinanderfolgenden Paar von Revisionen angewendet wurden, sodass ein Aufrufer die vollständige Änderungshistorie wiedergeben oder prüfen kann.
- get_revision_content(filename: str, revision: int) str[Quelle]#
Gibt den exakten Inhalt zurück, der in revision gespeichert ist.
Wenn die Revision nicht existiert, wird ein leerer String zurückgegeben, sodass nachgelagerter Code den Fall „nicht gefunden“ ohne Ausnahmen behandeln kann.
- get_revision_diffs(filename: str) List[str][Quelle]#
Gibt eine chronologische Liste von Unified-Diffs für filename zurück.
Jedes Element in der zurückgegebenen Liste repräsentiert den Diff, der Revision n in Revision n+1 umgewandelt hat (beginnend mit Revision 1 → 2).
- list_files() Dict[str, int][Quelle]#
Gibt eine Zuordnung von Dateiname → neueste Revisionsnummer zurück.
- get_latest_content(filename: str) str[Quelle]#
Gibt den aktuellsten Inhalt zurück oder einen leeren String, wenn die Datei neu ist.
- add_or_update_file(filename: str, new_content: str | bytes | Any) None[Quelle]#
Erstellt filename oder fügt eine neue Revision mit new_content hinzu.
- get_diff(filename: str, from_revision: int, to_revision: int) str[Quelle]#
Gibt einen Unified-Diff zwischen from_revision und to_revision zurück.
- class TextCanvasMemory(canvas: TextCanvas | None = None)[Quelle]#
Bases:
MemoryEine Speicherimplementierung, die einen Canvas zur Speicherung von dateiähnlichem Inhalt verwendet. Fügt den aktuellen Zustand des Canvas bei jeder Runde in den ChatCompletionContext ein.
Warnung
Dies ist eine experimentelle API und kann sich in Zukunft ändern.
Die TextCanvasMemory bietet einen persistenten, dateiähnlichen Speicherungsmechanismus, der von Agenten zum Lesen und Schreiben von Inhalten verwendet werden kann. Sie injiziert automatisch den aktuellen Zustand aller Dateien im Canvas in den Modellkontext vor jeder Inferenz.
Dies ist besonders nützlich für: - Ermöglichen von Agenten, Dokumente über mehrere Runden hinweg zu erstellen und zu ändern - Ermöglichen von kollaborativer Dokumentenbearbeitung zwischen mehreren Agenten - Aufrechterhalten eines persistenten Zustands über Gesprächsrunden hinweg - Arbeiten mit Inhalten, die zu groß sind, um in eine einzelne Nachricht zu passen
Der Canvas bietet Werkzeuge für: - Erstellen oder Aktualisieren von Dateien mit neuem Inhalt - Anwenden von Patches (im Unified-Diff-Format) auf bestehende Dateien
Examples
Beispiel: TextCanvasMemory mit einem AssistantAgent verwenden
Das folgende Beispiel zeigt, wie man eine TextCanvasMemory erstellt und sie mit einem AssistantAgent verwendet, um eine Story-Datei zu schreiben und zu aktualisieren.
import asyncio from autogen_core import CancellationToken from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.messages import TextMessage from autogen_ext.memory.canvas import TextCanvasMemory async def main(): # Create a model client model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) # Create the canvas memory text_canvas_memory = TextCanvasMemory() # Get tools for working with the canvas update_file_tool = text_canvas_memory.get_update_file_tool() apply_patch_tool = text_canvas_memory.get_apply_patch_tool() # Create an agent with the canvas memory and tools writer_agent = AssistantAgent( name="Writer", model_client=model_client, description="A writer agent that creates and updates stories.", system_message=''' You are a Writer Agent. Your focus is to generate a story based on the user's request. Instructions for using the canvas: - The story should be stored on the canvas in a file named "story.md". - If "story.md" does not exist, create it by calling the 'update_file' tool. - If "story.md" already exists, generate a unified diff (patch) from the current content to the new version, and call the 'apply_patch' tool to apply the changes. IMPORTANT: Do not include the full story text in your chat messages. Only write the story content to the canvas using the tools. ''', tools=[update_file_tool, apply_patch_tool], memory=[text_canvas_memory], ) # Send a message to the agent await writer_agent.on_messages( [TextMessage(content="Write a short story about a bunny and a sunflower.", source="user")], CancellationToken(), ) # Retrieve the content from the canvas story_content = text_canvas_memory.canvas.get_latest_content("story.md") print("Story content from canvas:") print(story_content) if __name__ == "__main__": asyncio.run(main())
Beispiel: TextCanvasMemory mit mehreren Agenten verwenden
Das folgende Beispiel zeigt, wie TextCanvasMemory mit mehreren Agenten verwendet wird, die an demselben Dokument zusammenarbeiten.
import asyncio from autogen_ext.models.openai import OpenAIChatCompletionClient from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.conditions import TextMentionTermination from autogen_ext.memory.canvas import TextCanvasMemory async def main(): # Create a model client model_client = OpenAIChatCompletionClient( model="gpt-4o", # api_key = "your_openai_api_key" ) # Create the shared canvas memory text_canvas_memory = TextCanvasMemory() update_file_tool = text_canvas_memory.get_update_file_tool() apply_patch_tool = text_canvas_memory.get_apply_patch_tool() # Create a writer agent writer_agent = AssistantAgent( name="Writer", model_client=model_client, description="A writer agent that creates stories.", system_message="You write children's stories on the canvas in story.md.", tools=[update_file_tool, apply_patch_tool], memory=[text_canvas_memory], ) # Create a critique agent critique_agent = AssistantAgent( name="Critique", model_client=model_client, description="A critique agent that provides feedback on stories.", system_message="You review the story.md file and provide constructive feedback.", memory=[text_canvas_memory], ) # Create a team with both agents team = RoundRobinGroupChat( participants=[writer_agent, critique_agent], termination_condition=TextMentionTermination("TERMINATE"), max_turns=10, ) # Run the team on a task await team.run(task="Create a children's book about a bunny and a sunflower") # Get the final story story = text_canvas_memory.canvas.get_latest_content("story.md") print(story) if __name__ == "__main__": asyncio.run(main())
- async update_context(model_context: ChatCompletionContext) UpdateContextResult[Quelle]#
Injiziert die gesamte Canvas-Zusammenfassung (oder eine ausgewählte Teilmenge) als Referenzdaten. Hier wird sie nur in eine Systemnachricht eingefügt, aber Sie könnten dies anpassen.
- async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) MemoryQueryResult[Quelle]#
Potenziell nach übereinstimmenden Dateinamen oder Dateiinhalt suchen. Dieses Beispiel gibt leer zurück.
- async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) None[Quelle]#
Beispielverwendung: Möglicherweise Inhalt als Patch oder direkte Dateiaktualisierung interpretieren. Könnte auch durch ein spezialisiertes „CanvasTool“ erfolgen.
- async clear() None[Quelle]#
Löscht den gesamten Canvas, indem er durch eine neue leere Instanz ersetzt wird.