NeMo-Guardrails中实现自定义动作的流式响应输出

在构建对话系统时，流式响应(Streaming Response)能够显著提升用户体验，使对话更加自然流畅。本文将详细介绍如何在NeMo-Guardrails框架中为自定义动作实现流式输出功能。

流式响应的核心原理

流式响应的核心在于逐步发送生成的内容，而不是等待整个响应完成后再一次性发送。在NeMo-Guardrails中，这一功能通过回调机制实现，允许LLM(大语言模型)在生成每个token时立即将其发送到客户端。

实现步骤详解

1. 获取流式处理句柄

NeMo-Guardrails内部维护了一个流式处理句柄(streaming handler)，可以通过streaming_handler_var.get()方法获取。这个句柄负责管理内容的流式传输。

2. 配置LLM调用参数

在调用LLM时，需要将流式处理句柄作为回调函数传入。这通过LangChain的RunnableConfig实现：

from langchain_core.runnables.config import RunnableConfig

call_config = RunnableConfig(callbacks=[streaming_handler_var.get()])

3. 异步调用LLM

使用异步调用方式(ainvoke)来触发LLM生成内容，并传入配置参数：

response = await llm.ainvoke(user_query, config=call_config)

4. 完整示例代码

下面是一个完整的自定义动作实现示例：

@action(is_system_action=True)
async def call_llm(user_query: str, llm: Optional[BaseLLM]) -> str:
    # 获取流式处理句柄
    handler = streaming_handler_var.get()
    
    # 配置回调
    call_config = RunnableConfig(callbacks=[handler])
    
    # 异步调用LLM并启用流式
    response = await llm.ainvoke(user_query, config=call_config)
    
    # 返回最终完整响应
    return response.content

技术要点说明

1. 异步处理：必须使用async/await语法，确保不阻塞事件循环。

2. 回调机制：流式处理依赖于LangChain的回调系统，每个生成的token都会触发回调。

3. 返回值处理：虽然实现了流式输出，但仍需返回完整内容供后续处理。

应用场景

这种流式输出技术特别适用于：

• 长文本生成场景
• 实时对话系统
• 需要快速反馈的用户界面

注意事项

1. 确保LLM本身支持流式输出
2. 客户端需要能够处理流式响应
3. 错误处理需要考虑流式场景下的特殊情况

通过上述方法，开发者可以轻松地在NeMo-Guardrails中为自定义动作添加流式输出功能，显著提升对话系统的用户体验。