Agently Instant模型返回结果流式解析

Agently Instant 是3.4.0版本引入的一项重要功能，为什么它的加入值得更新一个第二位大版本号？下面进行具体的介绍说明：

Note

⚠️ 为避免与OpenAI Realtime混淆，本模式自v3.4.0.4版本已经从Realtime更名为Instant，所有相关命名和用法中涉及"realtime"字样都已经统一更换为"instant"！ 更新记录
v3.4.0.1：在instant事件数据中加入delta字段，能够实时输出str类型的数据更新情况
v3.4.0.2：全面升级instant事件，支持对字典、列表类型相互嵌套的复杂的数据结构进行实时解析
v3.4.0.3：加入generator输出模式，让开发者能够通过for循环轮询的方式使用流式输出，同时简化了instant模式的声明方式，在代码中声明过instant相关的事件监听或是获取了instant generator都将自动启动instant模式，开发者再也不同担心忘记声明.use_instant()啦
v3.4.0.4：加入只监听特定key，indexes组合的generator输出模式，加入对使用&字符拼接多个监听条件的表达式支持，将原模式名称realtime改名为instant，以避免和OpenAI Realtime混淆误解，旧版realtime表达已经做了兼容处理

流式输出和结构化数据输出的矛盾

在我们进行基于语言模型进行应用开发时，在单次请求中，进行流式输出或是结构化数据（如JSON）输出，二者之间往往只能选其一。其实原因也很好理解，在流式输出过程中，模型不可能生成标准、封闭的结构化数据，再加上部分模型在生成结构化数据内容前后，还可能加上对生成内容的解释说明或是装饰符，这些情况，都指向了一个结论：我们很难在流式输出过程中，对结构化数据进行解析，只能等待模型给出完整输出的结束信号后，才能获得相对可靠、安全、完整的结构化数据。

但是，经过 Agently 团队的持续研究，加上开源社区中相关解决方案的贡献，再结合 Agently 框架原有的架构积累，我们能够自豪地推出 Agently Instant 方案，结合Agently特有的Output语法（点击这里了解更多），就能帮助开发者在模型流式输出的同时，第一时间获取并使用复杂数据结构中几乎任何一个特定字段的实时更新内容。是的，是几乎任何一个特定字段，包括从{ "list_a": [ { "dict_key_1": [ <str_item> ], ... }, ... ] }这样的复杂数据结构中，取出<str_item>的流式更新值。

对结构化数据输出进行流式、实时解析带来的效率提升

使用 Agently AI应用开发框架 的开发者们，可能都已经对结构化数据输出的好处非常熟悉，除了能够符合代码开发直觉地与代码中的各种变量、方法、模块进行通讯调用（点击这里了解更多）之外，使用结构化数据在单次请求中进行复杂的思维链规划、复杂条件输出、长文结构管理（点击这里了解更多）也是非常好用的语言模型输出控制方法，更不用说利用这个框架特性和Agently Workflow相结合，进行更复杂的工作流逻辑规划（点击这里了解更多）了。

但是，在3.4.0版本之前，要使用框架提供的便利地结构化数据输出方法来控制模型输出，开发者就只能在代码执行过程中等待完整的结构化数据输出，这无疑会降低整个代码执行的效率。下面，让我们用一个简单的实验设计，展示不同输出方案对同样命题的输出时长的影响：

>>>点击查看具体实验代码及实验过程输出结果记录

阶段	串行模式	并发模式	*Agently Instant* + 并发模式
生成3个单词并使用它们生成一句话	启动时间：0.00 分段耗时：3.27 完成时总耗时：3.27	启动时间：0.00 分段耗时：3.90 完成时总耗时：3.90	启动时间：0.00 分段耗时：3.23 完成时总耗时：3.23
使用第1个单词生成一句话	启动时间：3.27 分段耗时：2.39 完成时总耗时：5.66	启动时间：3.90 分段耗时：1.60 完成时总耗时：5.50	启动时间：1.66 分段耗时：3.59 完成时总耗时：4.67
使用第2个单词生成一句话	启动时间：5.66 分段耗时：2.80 完成时总耗时：8.46	启动时间：3.92 分段耗时：1.61 完成时总耗时：5.54	启动时间：1.93 分段耗时：1.61 完成时总耗时：4.38
使用第3个单词生成一句话	启动时间：8.47 分段耗时：1.50 完成时总耗时：9.96	启动时间：3.94 分段耗时：2.13 完成时总耗时：6.74	启动时间：2.29 分段耗时：2.11 完成时总耗时：3.90
最终总耗时	9.96	6.74	4.67

注：以上数据单位为秒，根据测试代码在Python控制台输出结果保留小数点后两位如实记录，可能由于Python代码实际执行过程及计时差异，出现偏差。

由于实验时，模型生成的单词难易度有高低，造句的长短和复杂度也有差异，因此，我们还要对原始数据进行如下对齐处理：

3个模式都使用将四个阶段的分段耗时加总结果作为未优化的串行执行期望总耗时，作为参照标准值；
使用（关键过程节点发生时的时点耗时/串行执行期望总耗时）*100%所获得的结果，作为关键过程节点发生时，实际消耗期望总耗时的进度百分比，称为关键过程节点进度；
使用串行执行期望节点进度（后简称期望进度）-关键过程节点进度（后简称实际进度）的差值，作为效率提升的参考值。

得到如下结果：

阶段	串行模式	并发模式	*Agently Instant* + 并发模式
生成3个单词并使用它们生成一句话	完成时期望进度：32.83% 完成时实际进度：32.83%	完成时期望进度：42.21% 完成时实际进度：42.21%	完成时期望进度：30.65% 完成时实际进度：30.65%
使用第1个单词生成一句话	开始时期望进度：32.83% 开始时实际进度：32.83% 完成时期望进度：56.93% 完成时实际进度：56.93%	开始时期望进度：42.21% 开始时实际进度：42.21% 完成时期望进度：59.52% 完成时实际进度：59.63%	开始时期望进度：30.65% 开始时实际进度：15.75% 完成时期望进度：76.95% 完成时实际进度：44.31%
使用第2个单词生成一句话	开始时期望进度：56.93% 开始时实际进度：56.93% 完成时期望进度：84.94% 完成时实际进度：84.94%	开始时期望进度：59.52% 开始时实际进度：42.42% 完成时期望进度：76.95% 完成时实际进度：59.96%	开始时期望进度：64.71% 开始时实际进度：18.31% 完成时期望进度：79.98% 完成时实际进度：41.56%
使用第3个单词生成一句话	开始时期望进度：84.94% 开始时实际进度：84.94% 完成时期望进度：100% 完成时实际进度：100%	开始时期望进度：76.95% 开始时实际进度：42.64% 完成时期望进度：100% 完成时实际进度：72.94%	开始时期望进度：79.98% 开始时实际进度：21.73% 完成时期望进度：100% 完成时实际进度：44.31%

进一步计算出：

阶段	串行模式	并发模式	*Agently Instant* + 并发模式
生成3个单词并使用它们生成一句话	完成时效率提升：0%	完成时效率提升：0%	完成时效率提升：0%
使用第1个单词生成一句话	开始时效率提升：0% 完成时效率提升：0%	开始时效率提升：0% 完成时效率提升：-0.11%	开始时效率提升：14.90% 完成时效率提升：32.64%
使用第2个单词生成一句话	开始时效率提升：0% 完成时效率提升：0%	开始时效率提升：17.10% 完成时效率提升：16.99%	开始时效率提升：46.40% 完成时效率提升：38.42%
使用第3个单词生成一句话	开始时效率提升：0% 完成时效率提升：0%	开始时效率提升：34.31% 完成时效率提升：27.06%	开始时效率提升：58.25% 完成时效率提升：55.69%

由上面的实验测算结果可见，在上述常见业务逻辑模拟的实验设计中，以完成效率为参考，使用 Agently Instant 方案，能够相对基础的串行方案带来超过55%的效率提升，即使相对目前常见的并发模式方案，也能再带来超过25%的效率提升。可以想见，面对更复杂、更长、更多请求的业务链条，采用 Agently Instant 方案将带来的更多的叠加效率提升效果。

如何使用

基本使用方法

使用 Agently Instant 方案非常简单，在数据出口部分，我们对框架原有的事件监听器模块进行了扩展，就做到了在不对原有框架使用方式进行大规模改动的前提下，开发者就能方便地用上新的能力。这也符合我们一贯追求的对开发者简单易用的价值主张。下面的代码展示了 Agently Instant 方案的基本使用方法：

import datetime
import Agently
agent = (
    Agently.create_agent()
        #.set_setting(...)
)

# 使用监听器监听新引入的instant事件
@agent.on_event("instant")
def instant_handler(data):
    # 返回的事件数据结构：
    # `key`: <str> 当前正在输出的键（采用Agently Instant表达方法）
    # `indexes`: <list> 如果当前正在输出的键路径中存在数组，`indexes`里会提供当前输出
    #                   是路径中数组的第几项
    # `delta`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新每次添加的新内容
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`value`字段值一致
    # `value`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新当前已生成的全量值
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`delta`字段值一致
    # `complete_value`: <any> 在当前事件抛出时，已经输出的结构化数据的全量内容

    # 输出Instant模式过程结果和输出时间
    print(datetime.now(), data["key"], data["indexes"], data["delta"])

result = (
    agent
        # 使用.use_instant()开启instant模式
        # 3.4.0.3版本之后可以省去此步
        .use_instant()
        .input("Generate 3 other words, then use those 3 words to make a sentence, then generate 4 numbers.")
        # 使用Agently Output语法定义一个复杂结构数据
        .output({
            "words": [("str", )],
            "sentence": ("str", ),
            "numbers": [{ "value": ("int", ) }]
        })
        .start()
)
# 输出最终结果和完成时间
print(datetime.now(), result)

运行结果：

Instant模式输出：
2024-11-03 02:20:01.650752 words.[].$delta [0] cat
2024-11-03 02:20:01.831325 words.[].$delta [1] mouse
2024-11-03 02:20:01.835427 words.[] [0] cat
2024-11-03 02:20:01.849140 words.[].$delta [2] run
2024-11-03 02:20:01.850624 words.[] [1] mouse
2024-11-03 02:20:01.912867 words [] ['cat', 'mouse', 'run']
2024-11-03 02:20:01.913157 words.[] [2] run
2024-11-03 02:20:01.962901 sentence.$delta [] The
2024-11-03 02:20:01.980559 sentence.$delta []  cat
2024-11-03 02:20:01.998184 sentence.$delta []  chased
2024-11-03 02:20:02.015376 sentence.$delta []  the
2024-11-03 02:20:02.032466 sentence.$delta []  mouse
2024-11-03 02:20:02.050336 sentence.$delta []  as
2024-11-03 02:20:02.088583 sentence.$delta []  it
2024-11-03 02:20:02.091482 sentence.$delta []  ran
2024-11-03 02:20:02.102013 sentence.$delta []  for
2024-11-03 02:20:02.118886 sentence.$delta []  its
2024-11-03 02:20:02.136612 sentence.$delta []  life
2024-11-03 02:20:02.154099 sentence.$delta [] .
2024-11-03 02:20:02.258635 sentence [] The cat chased the mouse as it ran for its life.
2024-11-03 02:20:02.556008 numbers.[] [0] {'value': 123}
2024-11-03 02:20:02.556662 numbers.[].value [0] 123
2024-11-03 02:20:02.747380 numbers.[] [1] {'value': 456}
2024-11-03 02:20:02.748144 numbers.[].value [1] 456
2024-11-03 02:20:02.938182 numbers.[] [2] {'value': 789}
2024-11-03 02:20:02.938688 numbers.[].value [2] 789
2024-11-03 02:20:03.483925  [] {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}
2024-11-03 02:20:03.484688 numbers [] [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]
2024-11-03 02:20:03.485579 numbers.[] [3] {'value': 101112}
2024-11-03 02:20:03.486465 numbers.[].value [3] 101112

最终Result：
2024-11-03 02:20:03.490869 {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}

Agently Instant模式的当前键表达语法

看过上面案例的输出结果，您一定已经注意到，我们采用了一套特殊的语法来表达当前输出的内容是来自于复杂数据结构中的哪一个键。下面是这套表达语法的一些基础规则的总结：

数据的基础结构通常为字典（dict）或列表（list）；
key字段采用.符号对字典键值层级进行分割；
key字段采用.[]符号和indexes字段配合，声明当前输出内容为列表内的元素项目，indexes字段中的元素数量和key字段中包含的.[]符号数量一致，指示当前正在输出的是列表中的第几个元素；
当前输出键内容为字符串（str）类型时，通过delta字段（只输出本次增量内容）和value字段（输出当前已经生成的全量键内容）提供两种流式输出模式，能够在键内容未完全生成的情况下获得更实时的流式输出，当内容完全生成完毕后，会通过后缀.$complete进行标识，以方便开发者在不同场景进行使用；
当前输出键内容为非字符串类型时，会在当前键内容完全生成后进行输出；
当前输出键内容为列表中元素（即列表键名.[]）时，会在元素完全生成完毕后进行输出；
当前输出键内容为列表（即仅列表键名，不包含.[]）时，会在列表中全部元素完全生成完毕后进行输出。

下面的案例可以帮助您更好地理解复杂数据结构和key字段、indexes字段的关联关系：

# 下面复杂数据结构中，键值字符串即为`key`和`indexes`的值，用|进行分割
{
    "value_a": "value_a | []",
    "dict_a": {
        "key_1": "dict_a.key_1 | []",
        "list_in_dict_a": [
            "dict_a.list_in_dict_a.[] | [0]",
            "dict_a.list_in_dict_a.[] | [1]",
            ...
        ],
        "list_with_dict_in_dict_a": [
            {
                "key_2": "dict_a.list_with_dict_in_dict_a.[].key_2 | [0]"
            },
            ...
        ]
    },
    "list_a": [
        "list_a.[] | [0]",
        "list_a.[] | [1]",
        ...
    ],
    "list_b": [
        {
            "list_with_dict_in_list_b": [
                {
                    "key_3": "list_b.[].list_with_dict_in_list_b.[].key_3 | [0, 0]"
                },
                {
                    "key_3": "list_b.[].list_with_dict_in_list_b.[].key_3 | [0, 1]"
                },
                ...
            ]
        },
        {
            "list_with_dict_in_list_b": [
                {
                    "key_3": "list_b.[].list_with_dict_in_list_b.[].key_3 | [1, 0]"
                },
                {
                    "key_3": "list_b.[].list_with_dict_in_list_b.[].key_3 | [1, 1]"
                },
                ...
            ]
        },
    ]
}

通过理解上面的键表达语法，您就可以使用 Agently Instant 方案，通过在instant事件监听器中加入条件过滤的方式，更加实时地获取和处理目标键内容。

例如如果想要从上面的案例中获取key_3（假设它是个字符串类型的键）的实时更新内容，您只需要在监听器中这样写：

@agent.on_event("instant")
def instant_handler(data):
    if data["key"] == "list_b.[].list_with_dict_in_list_b.[].key_3":
        print(data["delta"])
    elif ...:
        ...
    else ...:
        ...

更进一步，如果仅仅想要list_b全部元素中，list_with_dict_in_list_b的第一个元素里的key_3的值，您只需要在监听器中这样写：

@agent.on_event("instant")
def instant_handler(data):
    if (
        data["key"] == "list_b.[].list_with_dict_in_list_b.[].key_3"
        and data["indexes"][1] == 0
    ):
        print(data["delta"])
    elif ...:
        ...
    else ...:
        ...

再简单点，直接使用事件监听器监听特定键

v3.4.0.3更新：看完上面的开发方法之后，可能有的开发者会提出这样的问题：如果我只关心特定的少数键值的监听，为什么还要处理instant事件抛出的所有数据？有没有更简单的定点监听表达方式？

当然有， Agently Instant 方案也为开发者提供了instant:<key_expression>的监听表达方式。

同样用上面的案例，获取key_3的实时更新内容，您还可以这样写：

@agent.on_event("instant:list_b.[].list_with_dict_in_list_b.[].key_3")
def instant_handler(data):
    print(data["delta"])

更进一步，如果仅仅想要list_b全部元素中，list_with_dict_in_list_b的第一个元素里的key_3的值，您还可以这样写：

@agent.on_event("instant:list_b.[].list_with_dict_in_list_b.[].key_3?_,0")

其中通过?分割key和indexes内容，在indexes内容中，如果需要输入多个元素定位要求，可以通过,进行分割，其中_（或者*）表示接受该位置的所有元素，您也可以通过(0|2|4)的方式表达接受该位置的多个元素。

v3.4.0.4更新：再进一步，如果您希望监听器同时处理多个条件，可以使用&对多个条件进行组合：

@agent.on_event("instant:value_a&dict_a.key_1&list_b.[].list_with_dict_in_list_b.[].key_3?_,0")

跟上行业开发习惯，用Generator也可以输出流式事件

当然了，在其他行业工具中，流式输出往往会结合Generator一起使用，比如Gradio就是一个典型例子，如果要使用流式更新，就需要向它传递可以被for循环进行逐项轮询的Generator实例。在v3.4.0.3版本的更新中，Agently Instant 也带来了适配Generator的输出方案，您只需要将.start()换成.get_instant_generator()即可获取到包含所有instant事件的Generator输出实例了，示例代码如下：

generator = (
    agent
        .input("Generator 10 sentences")
        .output({
            "sentences": ([("str", )]),
        })
        .get_instant_generator()
)

for item in generator:
    print(item["key"], item["delta"])

Generator也可以指定监听的key和indexes

v3.4.0.4更新：在监听器中只针对特定事件的监听，在Generator中也可以做到：

# 监听器表达
@agent.on_event("instant:value_a&dict_a.key_1&list_b.[].list_with_dict_in_list_b.[].key_3?_,0")
def handler(data):
    pass

# Generator表达
generator = agent.get_instant_keys_generator("instant:value_a&dict_a.key_1&list_b.[].list_with_dict_in_list_b.[].key_3?_,0")

for item in generator:
    print(item["key"], item["delta"])