autogen_ext.memory.canvas#

class TextCanvas[来源]#

基类：BaseCanvas

一个内存画布，存储具有完整修订历史的文本文件。

警告

这是一个实验性API，未来可能会发生变化。

除了原始的类似CRUD的操作，这个增强的实现增加了

apply_patch – 使用 unidiff 库应用补丁，以实现准确的区块应用和上下文行验证。
get_revision_content – 随机访问任何历史修订版本。
get_revision_diffs – 获取每对连续修订版本之间应用的差异列表，以便调用者可以重放或审计完整的更改历史记录。

get_revision_content(filename: str, revision: int) → str[来源]#

返回存储在revision中的确切内容。

如果修订版本不存在，则返回一个空字符串，以便下游代码可以在没有异常的情况下处理“未找到”的情况。

get_revision_diffs(filename: str) → List[str][来源]#

返回filename的按时间顺序排列的统一差异列表。

返回列表中的每个元素表示将修订版本n转换为修订版本n+1的差异（从修订版本1 → 2开始）。

list_files() → Dict[str, int][来源]#: 返回filename → 最新修订版本号的映射。

get_latest_content(filename: str) → str[来源]#: 返回最新内容，如果文件是新文件则返回空字符串。

add_or_update_file(filename: str, new_content: str | bytes | Any) → None[来源]#: 创建filename或追加包含new_content的新修订版本。

get_diff(filename: str, from_revision: int, to_revision: int) → str[来源]#: 返回from_revision和to_revision之间的统一差异。

apply_patch(filename: str, patch_data: str | bytes | Any) → None[来源]#

将patch_text（统一差异）应用于最新修订版本并保存新的修订版本。

使用unidiff库准确应用区块并验证上下文行。

get_all_contents_for_context() → str[来源]#: 返回每个文件及其最新修订版本的摘要视图。

class TextCanvasMemory(canvas: TextCanvas | None = None)[来源]#

基类：Memory

一个使用 Canvas 存储文件类内容的内存实现。在每个回合中，将 Canvas 的当前状态插入到 ChatCompletionContext 中。

警告

这是一个实验性API，未来可能会发生变化。

TextCanvasMemory 提供了一个持久的、文件式的存储机制，可供代理用于读取和写入内容。它在每次推理之前自动将 Canvas 中所有文件的当前状态注入到模型上下文中。

这对于以下情况特别有用： - 允许代理在多个回合中创建和修改文档 - 启用多个代理之间的协作文档编辑 - 在会话回合之间维护持久状态 - 处理太大而无法放入单个消息的内容

Canvas 提供了以下工具： - 使用新内容创建或更新文件 - 对现有文件应用补丁（统一差异格式）

示例

示例：将 TextCanvasMemory 与 AssistantAgent 一起使用

以下示例演示了如何创建 TextCanvasMemory 并将其与 AssistantAgent 一起使用，以写入和更新故事文件。

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())

示例：将 TextCanvasMemory 与多个代理一起使用

以下示例展示了如何将 TextCanvasMemory 与在同一文档上协作的多个代理一起使用。

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[来源]#: 注入整个 canvas 摘要（或选定的子集）作为参考数据。这里，我们只是将其放入系统消息中，但您可以自定义。

async query(query: str | MemoryContent, cancellation_token: CancellationToken | None = None, **kwargs: Any) → MemoryQueryResult[来源]#: 可能搜索匹配的文件名或文件内容。此示例返回空值。

async add(content: MemoryContent, cancellation_token: CancellationToken | None = None) → None[来源]#: 示例用法：可能将内容解释为补丁或直接文件更新。也可以通过专门的“CanvasTool”完成。

async clear() → None[来源]#: 通过将其替换为新的空实例来清除整个 canvas。

async close() → None[来源]#: 清理内存实现使用的任何资源。

get_update_file_tool() → UpdateFileTool[来源]#: 返回一个与此内存 canvas 协同工作的 UpdateFileTool 实例。

get_apply_patch_tool() → ApplyPatchTool[来源]#: 返回一个与此内存 canvas 协同工作的 ApplyPatchTool 实例。