基于Langserve与Llamacpp的RAG应用流式响应问题求助

以下是针对你的代码的具体修改建议，解决终端能流式输出但Playground一次性返回的问题：

1. 修正PromptTemplate输入变量不匹配问题 #

你的Prompt模板使用了{context}和{question}作为变量，但model_response_prompt()函数返回的PromptTemplate却声明了错误的输入变量，这会导致变量绑定失败，先修正：

def model_response_prompt():
    return PromptTemplate(template=prompt, input_variables=['context', 'question'])

2. 替换同步回调为异步流式回调 #

原有的StreamingStdOutCallbackHandler是同步实现，无法适配LangServe的异步流式响应机制，改用异步版本：

# 替换导入语句
from langchain.callbacks.streaming_stdout import AsyncStreamingStdOutCallbackHandler

# 修改build_llm函数的默认callback参数
def build_llm(model_path, temperature=cfg.RAG_TEMPERATURE, max_tokens=cfg.MAX_TOKENS, callback = AsyncStreamingStdOutCallbackHandler()):
    callback_manager = CallbackManager([callback])
    # ... 其余代码保持不变
    llm = LlamaCpp(
        # ... 其余参数保持不变
        use_async=True,  # 添加异步支持
        streaming=True,
        # ...
    )
    return llm

3. 启用向量检索的异步模式 #

向量检索的同步操作会阻塞流式输出，将检索器改为异步模式：

# 修改vector_store.as_retriever的调用
context=(itemgetter("question") | vector_store.as_retriever(asynchronous=True)),

4. 调整Chain结构以适配流式输出 #

原Chain通过RunnableParallel同时返回answer和docs，这会导致流式输出被阻塞。如果只需要流式返回回答，可简化Chain结构：

final_chain = (
    RunnableParallel(
        context=(itemgetter("question") | vector_store.as_retriever(asynchronous=True)),
        question=itemgetter("question")
    )
    | prompt_temp
    | llm
).with_types(input_type=RagInput)

如果需要同时返回检索到的文档和流式回答，可使用RunnableMap并配置LangServe识别流式字段，但需要额外处理，建议先实现基础流式输出后再扩展。

5. 配置LangServe启用流式路由 #

在add_routes时明确指定流式支持：

add_routes(app, final_chain, path="/rag", streaming=True)

修改后的完整核心代码片段 #

# 修正后的PromptTemplate定义
def model_response_prompt():
    return PromptTemplate(template=prompt, input_variables=['context', 'question'])
prompt_temp = model_response_prompt()

# 异步回调和LLM构建
from langchain.callbacks.streaming_stdout import AsyncStreamingStdOutCallbackHandler
def build_llm(model_path, temperature=cfg.RAG_TEMPERATURE, max_tokens=cfg.MAX_TOKENS, callback = AsyncStreamingStdOutCallbackHandler()):
        callback_manager = CallbackManager([callback])
        n_gpu_layers = 1
        n_batch = 512
        llm = LlamaCpp(
                max_tokens = max_tokens,
                n_threads = 8,
                model_path=model_path,
                temperature=temperature,
                f16_kv=True,
                n_ctx=15000,
                n_gpu_layers=n_gpu_layers,
                n_batch=n_batch,
                callback_manager=callback_manager, 
                verbose=True,
                top_p=0.75,
                top_k=40,
                repeat_penalty = 1.1,
                streaming=True,
                use_async=True,  # 新增异步支持
                model_kwargs={},
        )
        return llm

# 异步检索器
final_chain = (
    RunnableParallel(
        context=(itemgetter("question") | vector_store.as_retriever(asynchronous=True)),
        question=itemgetter("question")
    )
    | prompt_temp
    | llm
).with_types(input_type=RagInput)

# 启用流式路由
add_routes(app, final_chain, path="/rag", streaming=True)

修改完成后重启服务，访问Playground即可看到流式输出效果。

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