将自定义流输出添加到代理

Overview

您可以配置自定义流输出，以指示代理除了纯文本之外，还为流媒体调用发出结构化的机器可读输出。 Atlas Agent Engine 将此输出作为自定义事件发送到客户端。

本指南介绍如何执行以下任务：

• 启用自定义流输出：在 agent.yaml文件中设置 features.use_custom_parser: true 标志。

• 注册输出解析器：在您的应用上注册一个 OutputParser 子类。

• 发出自定义事件：从代理代码中发出自定义事件。

先决条件

在开始之前，请确保您有一个带有 agent.yaml文件的注册代理。要学习；了解更多信息，请参阅Atlas助手引擎入门。

启用自定义流输出

要为代理启用自定义流输出，请将以下代码添加到 agent.yaml文件中：

features:
  use_custom_parser: true

如果将 features.use_custom_parser 标志设立为 true，但未注册输出解析器，则代理在调用启动时失败并返回RuntimeError。

注册输出解析器

使用 @app.output_parser 装饰器在您的应用上注册 OutputParser 子类。 Atlas助手引擎会在修饰时验证该类，因此在导入模块时，传递不是 OutputParser 子类的类会失败。

对于 LangGraph 代理，将 LangGraphOutputParser 类进行子类化。将 stream_modes 类属性设置为要转换为自定义事件的 LangGraph流模式。然后，实现parse() 和 on_stream_error() 方法。

以下示例为 LangGraph代理注册一个输出解析器：

from agent_engine_sdk_langgraph.output_parser import LangGraphOutputParser
@app.output_parser
class BriefParser(LangGraphOutputParser):
    stream_modes = ("messages", "values")
    async def parse(self, item, ctx):
        ...
        yield {"event": "brief", "data": new_brief}
    async def on_stream_error(self, ctx, error):
        ...
        return {"event": "error"}

下表描述了 LangGraphOutputParser 成员：

如果 parse() 或 on_stream_error() 方法引发错误，则流将失败，而不是默默地删除输出。

发出自定义事件

在流媒体传输调用期间，您可以使用 emit_custom_event() 函数直接从 LangGraph 节点、工具或在执行中运行的任何其他代码发出自定义事件。将包含字符串键的JSON对象传递给 emit_custom_event()，如以下示例所示：

from runner_shared import emit_custom_event
await emit_custom_event({"event": "todo", "items": ["a", "b"]})

Atlas助手引擎会按原样转发JSON对象，并且不会强加模式，因此您可以定义有效负载的结构以匹配客户端的期望。

调用 emit_custom_event() 函数时，应用以下规则：

• 仅通过流媒体调用调用该方法。从同步 invoke 路径调用它会引发错误。

• 启用 agent.yaml文件中的 features.use_custom_parser 标志。如果该标志被禁用，平台会引发错误。

• 异步代码中的 Await 方法。在同步代码中，请改用 emit_custom_event_sync()。

后续步骤

要学习；了解客户端如何接收流式输出，请参阅《调用代理》指南中的流式传输代理的输出。