autogen_ext.memory.canvas#

class TextCanvas[源代码]#

基类：BaseCanvas

一个内存中的画布，用于存储带有完整修订历史记录的文本文件。

警告

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

除了原始的类似 CRUD 的操作之外，此增强实现还添加了

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

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

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

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

使用unidiff库来准确地应用 hunks 并验证上下文行。

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

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

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

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][source]#: 返回一个 文件名 → 最新修订号 的映射。

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

基类：Memory

一种使用 Canvas 存储类似文件内容的内存实现。在每次轮询时将 canvas 的当前状态插入到 ChatCompletionContext 中。

警告

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

TextCanvasMemory 提供了一种持久的、类似文件的存储机制，代理可以使用它来读取和写入内容。它会自动将 canvas 中所有文件的当前状态注入到模型上下文中，然后再进行每次推断。

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

canvas 提供了以下工具： - 使用新内容创建或更新文件 - 将补丁（统一 diff 格式）应用于现有文件

示例

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

async clear() → None[source]#: 通过将整个画布替换为新的空实例来清除它。

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

get_apply_patch_tool() → ApplyPatchTool[source]#: 返回一个与此内存的 canvas 一起使用的 ApplyPatchTool 实例。

get_update_file_tool() → UpdateFileTool[source]#: 返回一个与此内存的 canvas 一起使用的 UpdateFileTool 实例。

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

async update_context(model_context: ChatCompletionContext) → UpdateContextResult[source]#: 将整个 canvas 摘要（或选定的子集）作为参考数据注入。在这里，我们只是将其放入系统消息中，但您可以自定义。