当前位置: 首页 > article >正文

Autogen_core: ClosureAgent使用与测试

目录

      • 第一个示例
      • 第二个示例
      • 完成的功能

下面两个示例展示了如何使用 AutoGen 库中的 ClosureAgent 来创建和使用代理。 ClosureAgent 允许你使用闭包(即一个没有定义类的函数)来定义代理,并从运行时中提取值。代码中展示了两个示例:

第一个示例

import asyncio
from dataclasses import dataclass

from autogen_core import (
    ClosureAgent,
    ClosureContext,
    DefaultSubscription,
    DefaultTopicId,
    MessageContext,
    SingleThreadedAgentRuntime,
)
"""
The closure agent allows you to define an agent using a closure, or function without needing to define a class. It allows values to be extracted out of the runtime.

        The closure can define the type of message which is expected, or `Any` can be used to accept any type of message.

"""
@dataclass
class MyMessage:
    content: str


async def main():
    queue = asyncio.Queue[MyMessage]()

    async def output_result(_ctx: ClosureContext, message: MyMessage, ctx: MessageContext) -> None:
        await queue.put(message)

    runtime = SingleThreadedAgentRuntime()
    await ClosureAgent.register_closure(
        runtime, "output_result", output_result, subscriptions=lambda: [DefaultSubscription()]
    )

    runtime.start()
    await runtime.publish_message(MyMessage("Hello, world!"), DefaultTopicId())
    await runtime.stop_when_idle()

    result = await queue.get()
    print(result)


await main()

MyMessage(content='Hello, world!')

这个示例定义了一个简单的代理,它接收一个消息并将其放入一个 asyncio 队列中。

  1. 定义了一个名为 MyMessage 的数据类,用于存储消息内容。
  2. 创建了一个名为 output_result 的异步函数,它接收一个 ClosureContext、一个 MyMessage 和一个 MessageContext,然后将消息放入队列中。
  3. 创建了一个 SingleThreadedAgentRuntime 实例。
  4. 使用 register_closure 方法注册了 output_result 函数作为一个代理。它订阅了默认的主题。
  5. 启动运行时,发布一个 MyMessage 消息,并在空闲时停止运行时。
  6. 从队列中获取结果并打印。

这个示例展示了如何注册一个闭包代理,并在运行时中发送和接收消息。

第二个示例

@dataclass
class Message:
    content: str


async def test_register_receives_publish() -> None:
    runtime = SingleThreadedAgentRuntime()

    queue = asyncio.Queue[tuple[str, str]]()

    async def log_message(closure_ctx: ClosureContext, message: Message, ctx: MessageContext) -> None:
        key = closure_ctx.id.key
        await queue.put((key, message.content))

    await ClosureAgent.register_closure(runtime, "name", log_message, subscriptions=lambda: [DefaultSubscription()])
    runtime.start()

    await runtime.publish_message(Message("first message"), topic_id=DefaultTopicId())
    await runtime.publish_message(Message("second message"), topic_id=DefaultTopicId())
    await runtime.publish_message(Message("third message"), topic_id=DefaultTopicId())

    await runtime.stop_when_idle()

    assert queue.qsize() == 3
    assert queue.get_nowait() == ("default", "first message")
    assert queue.get_nowait() == ("default", "second message")
    assert queue.get_nowait() == ("default", "third message")
    assert queue.empty()
    
test_register_receives_publish()
<coroutine object test_register_receives_publish at 0x00000150ABEAA640>

这个示例定义了一个测试函数,用于测试注册代理、接收和发布消息的功能。

  1. 定义了一个名为 Message 的数据类,用于存储消息内容。
  2. 创建了一个名为 test_register_receives_publish 的异步测试函数。
  3. 在这个函数中,创建了一个 SingleThreadedAgentRuntime 实例和一个 asyncio 队列。
  4. 定义了一个名为 log_message 的异步函数,它接收一个 ClosureContext、一个 Message 和一个 MessageContext,然后将代理的 ID 和消息内容作为一个元组放入队列中。
  5. 使用 register_closure 方法注册了 log_message 函数作为一个代理,并订阅了默认的主题。
  6. 启动运行时,并发布三个 Message 消息。
  7. 等待运行时空闲后,检查队列中的消息数量和内容,确保有三个消息,并且它们的顺序和内容与发布的消息一致。

这个示例展示了如何测试代理的注册、消息的接收和发布功能。

完成的功能

这两个示例展示了如何使用 ClosureAgentSingleThreadedAgentRuntime 来创建、注册和使用闭包代理。它们展示了如何发送和接收消息,以及如何测试代理的功能。这些示例是理解和学习如何使用 AutoGen 库来构建代理和消息传递系统的基础。

原文地址:https://blog.csdn.net/qq_41472205/article/details/145400326
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.kler.cn/a/526126.html

相关文章:

  • selenium定位网页元素
  • 如何使用深度学习中的 Transformer 算法进行视频目标检测
  • C基础寒假练习(4)
  • 【Rust自学】17.3. 实现面向对象的设计模式
  • MSU:通过图结构增强LLM推理
  • Vue3的el-table-column下拉输入实时查询API数据选择的实现方法
  • 力扣【1049. 最后一块石头的重量 II】Java题解(背包问题)
  • Windows程序设计8:获取文件大小的两种方式
  • 【HarmonyOS之旅】基于ArkTS开发(三) -> 兼容JS的类Web开发(一)
  • Oracle Primavera P6 最新版 v24.12 更新 2/2
  • 数据结构 前缀中缀后缀
  • 毕业设计--具有车流量检测功能的智能交通灯设计
  • 【二叉树的深搜】二叉树剪枝
  • Ubuntu安装VMware17
  • C++ 堆栈分配的区别
  • 【Block总结】PConv,部分卷积|即插即用
  • 【数据结构】最有效的实现栈和队列的方式(CC++语言版)
  • 计算机组成原理学习笔记
  • 组合模式 - 组合模式的实现
  • 从0开始使用面对对象C语言搭建一个基于OLED的图形显示框架(OLED设备层封装)