如何用Vercel AI SDK流式输出LangGraph生成结果？报错求助

问题背景 #

需要基于Vercel AI SDK实现LangGraph的流式输出，已知该SDK与LangChain LCEL配合正常，但在LangGraph场景下调用LangChainAdapter.toDataStreamResponse(final_generation)时出现错误：TypeError: stream.pipeThrough is not a function。当前后端接口通过streamEvents收集LLM输出chunk，最后尝试转换为响应流时失败，且无法实现实时流式输出。

错误原因 #

1. LangChainAdapter.toDataStreamResponse接收的是LangChain标准流式迭代器，而当前代码传入的是单个chunk对象，不符合方法参数要求，导致pipeThrough报错。
2. 当前代码通过for await...of循环收集所有chunk后才返回响应，本质是一次性输出，没有实现真正的流式传输。

解决方案 #

1. 调整LangGraph节点的LLM调用逻辑 #

确保generate节点返回流式输出，而非完整字符串：

// generate节点修改后代码
import { State } from '$lib/server/graph/state';
import { LLMClient } from '$lib/server/llm-client';
import { ChatPromptTemplate } from '@langchain/core/prompts';

export const generate = async function* (state: typeof State.State) {
    console.log("---LLM Inference---");
    const PROMPT_TEMPLATE = 'You are a helpful assistant!';
    const prompt = ChatPromptTemplate.fromMessages([
        ['system', PROMPT_TEMPLATE],
        ['human', "{question}"],
    ]);
    // 使用stream方法获取流式迭代器，而非invoke
    const stream = await prompt.pipe(LLMClient.getClient()).stream({question: state.question});
    
    // 逐个yield chunk，让LangGraph能流式处理
    for await (const chunk of stream) {
        yield { generation: chunk.content };
    }
};

注意：将generate改为异步生成器函数（async function*），这样LangGraph会把每个chunk作为状态更新逐步输出，而非等待完整结果。

2. 修改后端API接口，实时输出流式响应 #

不再使用streamEvents收集chunk，而是直接通过LangGraph的stream方法获取状态流，创建符合Vercel AI要求的SSE响应：

// src/routes/api/chat 修改后代码
import type { RequestHandler } from './$types';
import type { Message } from 'ai/svelte';
import { Workflow } from '$lib/server/graph/workflow';

export const POST: RequestHandler = async ({ request }) => {
    const { messages }: { messages: Message[] } = await request.json();
    const inputQuestion = messages.pop()?.content as string;
    
    // 创建ReadableStream，将LangGraph的状态流转为SSE格式
    const stream = new ReadableStream({
        async start(controller) {
            const encoder = new TextEncoder();
            try {
                // 调用LangGraph的stream方法获取状态流
                const graphStream = await Workflow.getCompiledStateGraph().stream(
                    { question: inputQuestion, chat_history: messages },
                    { streamMode: "values" } // 只获取状态更新，而非事件
                );
                
                for await (const state of graphStream) {
                    if (state.generation) {
                        // 按照Vercel AI的SSE格式发送chunk
                        controller.enqueue(encoder.encode(`data: ${JSON.stringify({
                            id: Date.now().toString(),
                            content: state.generation,
                            role: 'assistant'
                        })}\n\n`));
                    }
                }
            } catch (err) {
                controller.error(err);
            } finally {
                controller.close();
            }
        }
    });
    
    // 返回SSE响应
    return new Response(stream, {
        headers: {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive'
        }
    });
};

3. 调整GraphState定义（可选） #

如果需要在状态中逐步拼接生成内容，可以调整generation的reducer：

// State定义修改后
import type { DocumentInterface } from '@langchain/core/documents';
import { Annotation, MessagesAnnotation } from '@langchain/langgraph';

export const GraphState = Annotation.Root({
    documents: Annotation<DocumentInterface[]>({
        reducer: (x, y) => y ?? x ?? []
    }),
    question: Annotation<string>({
        reducer: (x, y) => y ?? x ?? ''
    }),
    generation: Annotation<string>({
        // 逐步拼接chunk内容
        reducer: (x, y) => (x || '') + (y || ''),
        default: () => ''
    }),
    ...MessagesAnnotation.spec
});

关键说明 #

• 放弃使用LangChainAdapter.toDataStreamResponse，因为它仅适配LangChain LCEL的标准流，LangGraph的事件流/状态流格式不同，需要手动构建SSE响应。
• 通过异步生成器函数（async function*）让LangGraph节点支持流式输出，确保每个chunk能实时传递到前端。
• 前端的useChat()组件会自动处理SSE流，无需额外修改，即可实现实时渲染输出内容。

内容的提问来源于stack exchange，提问作者SparkOn