autogen_ext.models.replay

class ReplayChatCompletionClient(chat_completions: Sequence[str | CreateResult], model_info: ModelInfo | 无 = None)

基础类: ChatCompletionClient, Component[ReplayChatCompletionClientConfig]

一个模拟的聊天完成客户端，使用基于索引的方法回放预定义的响应。

该类通过预定义的响应列表来模拟聊天完成客户端。它支持单次完成和流式响应。响应可以是字符串或CreateResult对象。客户端现在使用基于索引的方法来访问响应，允许重置状态。

注意

响应可以是字符串或CreateResult对象。

Parameters:

chat_completions (Sequence[Union[str, CreateResult]]) – 用于回放的预定义响应列表。

Raises:

ValueError("No more mock responses available") – 如果提供的输出列表已用尽。

示例：

简单的聊天完成客户端，用于返回预定义的响应。

from autogen_core.models import UserMessage
from autogen_ext.models.replay import ReplayChatCompletionClient


async def example():
    chat_completions = [
        "Hello, how can I assist you today?",
        "I'm happy to help with any questions you have.",
        "Is there anything else I can assist you with?",
    ]
    client = ReplayChatCompletionClient(chat_completions)
    messages = [UserMessage(content="What can you do?", source="user")]
    response = await client.create(messages)
    print(response.content)  # Output: "Hello, how can I assist you today?"

简单的流式聊天完成客户端，用于返回预定义的响应

import asyncio
from autogen_core.models import UserMessage
from autogen_ext.models.replay import ReplayChatCompletionClient


async def example():
    chat_completions = [
        "Hello, how can I assist you today?",
        "I'm happy to help with any questions you have.",
        "Is there anything else I can assist you with?",
    ]
    client = ReplayChatCompletionClient(chat_completions)
    messages = [UserMessage(content="What can you do?", source="user")]

    async for token in client.create_stream(messages):
        print(token, end="")  # Output: "Hello, how can I assist you today?"

    async for token in client.create_stream(messages):
        print(token, end="")  # Output: "I'm happy to help with any questions you have."

    asyncio.run(example())

使用 .reset 重置聊天客户端状态

import asyncio
from autogen_core.models import UserMessage
from autogen_ext.models.replay import ReplayChatCompletionClient


async def example():
    chat_completions = [
        "Hello, how can I assist you today?",
    ]
    client = ReplayChatCompletionClient(chat_completions)
    messages = [UserMessage(content="What can you do?", source="user")]
    response = await client.create(messages)
    print(response.content)  # Output: "Hello, how can I assist you today?"

    response = await client.create(messages)  # Raises ValueError("No more mock responses available")

    client.reset()  # Reset the client state (current index of message and token usages)
    response = await client.create(messages)
    print(response.content)  # Output: "Hello, how can I assist you today?" again


asyncio.run(example())

classmethod _from_config(config: ReplayChatCompletionClientConfig) → 自我

从配置对象创建组件的新实例。

Parameters:

config (T) – 配置对象。

Returns:

Self – 组件的新实例。

_to_config() → ReplayChatCompletionClientConfig

导出配置，该配置将用于创建一个与此实例配置相匹配的组件新实例。

Returns:

T – 组件的配置。

actual_usage() → RequestUsage

property capabilities: ModelCapabilities

返回模拟功能。

async close() → 无

component_config_schema

别名 ReplayChatCompletionClientConfig

component_provider_override: ClassVar[str | 无] = 'autogen_ext.models.replay.ReplayChatCompletionClient'

覆盖组件的提供商字符串。这应用于防止内部模块名称成为模块名称的一部分。

component_type: ClassVar[ComponentType] = 'replay_chat_completion_client'

组件的逻辑类型。

count_tokens(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = []) → int

async create(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = [], json_output: bool | 无 = None, extra_create_args: 映射[str, 任何] = {}, cancellation_token: CancellationToken | 无 = None) → CreateResult

从列表中返回下一个完成项。

property create_calls: 列表[字典[str, 任何]]

返回调用create方法时的参数。

async create_stream(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = [], json_output: bool | 无 = None, extra_create_args: 映射[str, 任何] = {}, cancellation_token: CancellationToken | 无 = None) → AsyncGenerator[str | CreateResult, 无]

将下一个完成作为流返回。

property model_info: ModelInfo

remaining_tokens(messages: Sequence[已注解[系统消息 | UserMessage | AssistantMessage | FunctionExecutionResultMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]], *, tools: Sequence[工具 | 工具模式] = []) → int

reset() → 无

将客户端状态和使用情况重置为初始状态。

set_cached_bool_value(value: bool) → 无

total_usage() → RequestUsage