【流式输出】LangChain流式输出的概念

😁 作者简介：一名大四的学生，致力学习前端开发技术
⭐️个人主页：夜宵饽饽的主页
❔ 系列专栏：JavaScript小贴士
👐学习格言：成功不是终点，失败也并非末日，最重要的是继续前进的勇气

​🔥​前言：

流式输出是一种极具价值的技术，它允许数据在生成的同时逐步传输，而无需等待整个过程完成，这是我自己的学习笔记,希望可以帮助到大家，欢迎大家的补充和纠正

Streaming流式输出

单个LLM调用的运行时间通常比传统资源请求运行得更长，当构建需要多个推理步骤的复杂的链式或代理时，这种情况会更加复杂

为了解决这个问题，我们可以采用流式输出，输出一旦生成，就可以立即显示，在langchain框架中实现有三种方法

1、单个LLM的流式输出

在Langchain中，大多数模块都包含.stream方法作为处理流式输出的接口，.stream方法会返回一个迭代器，你可以将其与for await 一起使用...of循环

const azureModel=await getAzureModel()

const prompt=ChatPromptTemplate.fromTemplate(
    "告诉我一个关于{topice}的笑话"
)

const parse=new StringOutputParser()
const chain=prompt.pipe(azureModel).pipe(parse)

const eventStream=await chain.streamEvents(
    {topice:"猫"},
    {version:"v2"}
)

for await (const event of eventStream){
    const kind=event.event
    if(kind==="on_chat_model_stream"){
        console.log(event)
    }
}

2、从多个链式调用中获取中间结果

在langchain中我们可以使用streamEvents方法来实现，调用该方法时，它将返回一个迭代器，该迭代器会生成各种类型的事件，我们可以根据项目的需要筛选和处理这些事件

事件参考：

• on_llm_start：当大模型开始处理输入时触发该事件。输入内容是大模型的输入，没有输出内容
• on_llm_stream：当大模型逐步生成并流失传输输出是触发该事件，输出内容是部分输出结果
• on_llm_end：当大模型处理完成所有输入并生成完整输出是触发该事件，输出内容是完整的输出结果
• on_chain_start：当一个链式任务开始时触发该事件
• on_chain_stream：当链式任务的某一个步骤正在执行并产生输出时触发该事件
• on_chain_end：当链式任务完成时触发该事件
• on_tool_start：当工具开始开始执行时触发该事件
• on_tool_stream：当工具逐步生成并流式传输输出时触发该事件
• on_tool_end：当工具完成执行时触发该事件
• on_retriever_start:当检索器开始执行时触发该事件
• on_retriever_chunk：当检索器逐步返回结果时触发该事件
• on_retriever_end：当检索器完成检索任务时触发该事件
• on_prompt_start：当提示模版开始执行时触发该事件
• on_prompt_end：当提示模版完成时触发该事件

const azureModel=await getAzureModel()

const prompt=ChatPromptTemplate.fromTemplate(
    "告诉我一个关于{topice}的笑话"
)

const parse=new StringOutputParser()
const chain=prompt.pipe(azureModel).pipe(parse)

const eventStream=await chain.streamEvents(
    {topice:"猫"},
    {version:"v2"}
)

for await (const event of eventStream){
    const kind=event.event
    if(kind==="on_chat_model_stream"){
        console.log(event)
    }
}

3、自定义事件流

在有些时候，我们可能希望从Runnable中插入自定义回调事件，以便它可以显示在自定义回调处理程序中，例如：如果您有一个包含多个步骤的长时间运行的工具，则可以在步骤之间调度自定义事件，并使用这些自定义事件来监控进度。您还可以向应用程序的最终用户显示这些自定义事件，以向他们展示当前任务的进度。

要自定义事件，需要使用两个属性：name和data

name:字符串的格式，表示自定义的事件名称

data：任何格式，与事件关联的数据，这可以是任何东西，不过建议将器设置为JSON可序列化

//以下代码案例是需要环境支持async_hooks
const azureModel=await getAzureModel()

const reflect=RunnableLambda.from(async (value:string)=>{
    await dispatchCustomEvent("event1",{
        reversed:value.split("").reverse().join("")
    });
    await dispatchCustomEvent("events",5);
    return value
})

//--第一种方法使用eventStream来调用，这种方式更加适合需要实时处理事件的应用场景
const eventStream=await reflect.streamEvents("hello world",{
    version:"v2"
})

for await (const event of eventStream){
    // if(event.event==="on_custom_event"){
    //     console.log(event)
    // }
    console.log(event)
}

//--第二种方法使用回调函数来实现，适合短期的单次任务或调用
let result=await reflect.invoke("hello world",{
    callbacks:[
        {
            handleCustomEvent(eventName,data,runId){
                console.log(eventName,data,runId)
            }
        }
    ]
})

console.log(result)


//输出结果
/**
{
  event: 'on_custom_event',
  run_id: '9eac217d-3a2d-4563-a91f-3bd49bee4b3d',
  name: 'event1',
  tags: [],
  metadata: {},
  data: { reversed: 'dlrow olleh' }
}
{
  event: 'on_custom_event',
  run_id: '9eac217d-3a2d-4563-a91f-3bd49bee4b3d',
  name: 'event2',
  tags: [],
  metadata: {},
  data: 5
}
**/