how to use RunnableParallel

[shaojun]

My scenario is user always input multiple prompts and a single content, then in backend, each prompt + the single content should send to LLM separately and concurrently for responses.
I think the RunnableParallel is best match here, and it works until now I want to make it in langserve.

the sample code:

import uvicorn
from fastapi import FastAPI
from langchain.prompts import ChatPromptTemplate
# from langchain_community.chat_models import ChatOpenAI
from langserve import add_routes
from fastapi import FastAPI, File, UploadFile
import pathlib
from model_builder import ChatModelBuilder, ChatModelBuilderModelTypeEnum
from langchain.schema.runnable import RunnableLambda
from langchain_core.runnables import RunnableParallel
import json
import datetime
from langchain.prompts import ChatPromptTemplate
import os
from langserve import add_routes
from model_builder import ChatModelBuilder, ChatModelBuilderModelTypeEnum
from langserve import CustomUserType
from langchain_core.runnables import RunnableParallel, Runnable
from typing import Any, Dict, List, Tuple, Mapping
from langchain_openai import ChatOpenAI
app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple api server using Langchain's Runnable interfaces",
)

class MultiExpertsSummarizeRequest(CustomUserType):
    content: str
    expert_prompts: list[str]


class MultiExpertsSummarizeService:

    @staticmethod
    def process(request: MultiExpertsSummarizeRequest):
        prompts: list[ChatPromptTemplate] = []
        for expert_prompt in request.expert_prompts:
            prompts.append(ChatPromptTemplate.from_messages([
                ("system", expert_prompt),
                ("human", request.content)
            ]))
        model = ChatOpenAI()
        maps: Mapping[str, Runnable] = {}
        for index, expert_prompt in enumerate(prompts):
            expert_chain = expert_prompt | model
            maps.update({str(index): expert_chain})
        return maps


add_routes(
    app,
    RunnableLambda(MultiExpertsSummarizeService.process).with_types(
        input_type=MultiExpertsSummarizeRequest) | RunnableParallel(),
    path="/services/MultiExpertsSummarize",
)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=9000)

when I was playing at http://localhost:9000/services/MultiExpertsSummarize/playground/
there's always empty Output.
could you help for how to use it to satisfy my requirement(and make sure streaming interface are supported)?

[eyurtsev]

Why not use the batch API?

---

Bug in the code:

add_routes(
    app,
    RunnableLambda(MultiExpertsSummarizeService.process).with_types(
        input_type=MultiExpertsSummarizeRequest) | RunnableParallel(),
    path="/services/MultiExpertsSummarize",
)

Should be:

add_routes(
app,
RunnableLambda(MultiExpertsSummarizeService.process).with_types(
input_type=MultiExpertsSummarizeRequest),
path="/services/MultiExpertsSummarize",
)

[eyurtsev]

Marking as answer -- built in batch API is likely the way to solve the issue.

[shaojun]

does the batch API support streaming?