pass parameter betweens each Runnables in a chain

[shaojun]

Hi,
I have a CustomUserType input request, and there's a field of shared_pass_through_parameter that contains non-nature language data which would be used in any followed Runnables (except the LLM node).
like the below sample code, the purpose of this field is to pass to final RunnableLambda(output_process) to output a customized content.

app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="A simple api server using Langchain's Runnable interfaces",
)

# model = ChatModelBuilder.Create()


class MyCustomRequest(CustomUserType):
    content: str
    prompt: str
    shared_pass_through_parameter: str = "No-natural language content"


def process(request: MyCustomRequest):
    from pydantic import BaseModel, create_model
    import json
    prompt = ChatPromptTemplate.from_messages([
        ("system", '{system_prompt}'),
        ("human", '{user_content}')
    ])
    return prompt.invoke({"system_prompt": request.prompt, "user_content": request.content})


def output_process(input: Any):
    # !!!HOW TO GET the shared_pass_through_parameter
    # if input.shared_pass_through_parameter == 'test':
    #     print('shared_pass_through_parameter is test')
    return input


model = ChatOpenAI()
add_routes(
    app,
    RunnableLambda(process).with_types(
        input_type=MyCustomRequest) | model | RunnableLambda(output_process),
    path="/MyService1",
)
if __name__ == "__main__":
    import uvicorn

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

Obvious that field will be dropped as it's not included into the output of process.
I can see a solution that wrap all the route chain into a single function, but that will lose the feature of streaming of LLM.
any suggestion?

[eyurtsev]

@shaojun This looks like an LCEL question. There are a number of LCEL primitives that help manipulate dictionaries.

Here's an example that does something with dict manipulation:

https://github.com/langchain-ai/langserve/blob/main/examples/passthrough_dict/server.py

Look through LangChain docs to learn how to use PassthroughRunnable, PassthroughRunnable.assign, beta.runnables.context.Context to manipulate LCEL expressions

[shaojun]

@eyurtsev thanks for the reply.
I've played a while for the Context, but seems still could not work for my case that using async def function as streaming is a must feature.
My chain is doing like a translator with must:

1. every streamed response piece should be a json
   this is straigthfoward, and can be done via streaming and JsonOutputParser.
2. every streamed response piece(a incremental json) should append a json property which is the output from 1st runnable(a string extracted from a local file)
   the extra_info is the parameter generated at 1st runnable and I want it to passthrough all the chain.
   so the expecting data that client side continues seeing are like:
   1st:
   {"content":"你","extra_info": "i am from 1st runnable"}
   2nd:
   {"content":"你今","extra_info": "i am from 1st runnable"}
   3rd:
   {"content":"你今天","extra_info": "i am from 1st runnable"}
   ...
   ...

After some experiment, I see a problem in the bottom of the sample code.
in process_attach_extra_info(input_stream: Any) I can't get extra_info , so there's no way to keep appending extra_info (as a json property) to the incremental json, instead, an exception occured:

can see the output json is not full, the content is empty, and then got the exception:

...
...
File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\tracers\log_stream.py", line 237, in tap_output_aiter
async for chunk in output:
File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\runnables\base.py", line 2850, in _atransform
chunk = AddableDict({step_name: task.result()})
File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\runnables\base.py", line 2833, in get_next_chunk
return await py_anext(generator)
File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\runnables\base.py", line 1065, in atransform
final = final + chunk # type: ignore[operator]
TypeError: unsupported operand type(s) for +: 'dict' and 'dict'

My sample code:

model = ChatOpenAI()

class SimpleOutputFormat(CustomUserType):
    content: str = None

def process(request: str):
    # extra_info = file.read('a local file path')
    return extra_info 

def process_generate_prompt(request: str):
    parser = PydanticOutputParser(
        pydantic_object=SimpleOutputFormat)
    prompt = ChatPromptTemplate.from_messages([
        ("system", """you are a translator, can translate to chinese\n
{format_instructions}"""),
        ("human", '{user_input}')
    ])
    return prompt.invoke({"user_input": request, "format_instructions": parser.get_format_instructions()})


async def process_attach_extra_info(input_stream: Any):
    async for input in input_stream:
        if "extra_info" in input:
            print(
                "!!!this never get hit and instead poped up an exception. and if get hit, how to get the json object parsed from llm's output as the base for appending new property?")
            # extra_data = {"extra_info": input["extra_info"]}
            # llm_output_json_obj["extra_info"] = extra_data
            # yield llm_output_json_obj
        if "llm_model_output_stream" in input:
            print('this get hit just serveral times, and not full content')
            d = input["llm_model_output_stream"]
            yield d


add_routes(
    app,
    RunnableLambda(process) |
    Context.setter("extra_info") |
    RunnableLambda(process_generate_prompt) | model
    | JsonOutputParser()
    | {"llm_model_output_stream": RunnablePassthrough(), "extra_info": Context.getter("extra_info")}
    | process_attach_extra_info,
    path="/path2",
)

Could you help?

[eyurtsev]

TypeError: unsupported operand type(s) for +: 'dict' and 'dict' -> looks like a bug in langchain_core

If you're able to produce minimal code that re-creates it it'll be helpful to get a fix out

[shaojun]

@eyurtsev
simply say I want make the first runnable's output as a parameter can pass through the chain, and in the last runnable, can get this parameter to keep appending as a json property to the incremental json object.
so the expecting data that client side continues seeing are like:
1st:
{"content":"你","extra_info": "i am from 1st runnable"}
2nd:
{"content":"你今","extra_info": "i am from 1st runnable"}
3rd:
{"content":"你今天","extra_info": "i am from 1st runnable"}
...
...

as I want to output json, so the minimal code I can produce is:

import uuid
from langchain_core.beta.runnables.context import Context
model = ChatOpenAI()

class SimpleOutputFormat(CustomUserType):
    content: str = None


def process(request: str):
    return request+str(uuid.uuid4())


def process_generate_prompt(request: str):
    parser = PydanticOutputParser(
        pydantic_object=SimpleOutputFormat)
    prompt = ChatPromptTemplate.from_messages([
        ("system", """you are a translator, can translate to chinese\n
{format_instructions}"""),
        ("human", '{user_input}')
    ])
    return prompt.invoke({"user_input": request, "format_instructions": parser.get_format_instructions()})


async def process_attach_original_user_input(input_stream: Any):
    async for input in input_stream:
        # how to get the input_with_id and keep appending it as a json property to the incremental json output?
        yield input

add_routes(
    app,
    RunnableLambda(process) |
    Context.setter("input_with_id") |
    RunnableLambda(process_generate_prompt) | model
    | JsonOutputParser()
    | {"parser_output_stream": RunnablePassthrough(), "extra_info": Context.getter("input_with_id")},
    path="/path2",
)

then see exception:

File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\runnables\base.py", line 2833, in get_next_chunk
return await py_anext(generator)
File "c:\Users\music\miniconda3\envs\langserve\lib\site-packages\langchain_core\runnables\base.py", line 1065, in atransform
final = final + chunk # type: ignore[operator]
TypeError: unsupported operand type(s) for +: 'dict' and 'dict'

please take a look for to fix the exception, and achieve my goal of continus appending json property.

[eyurtsev]

Thank you i created an issue: #504

[eyurtsev]

OK first here's a simple solution without Context:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableGenerator
from langchain_openai.chat_models import ChatOpenAI

from langserve import CustomUserType


class MyCustomRequest(CustomUserType):
    content: str
    prompt: str
    shared_pass_through_parameter: str = "No-natural language content"


model = ChatOpenAI()

prompt = ChatPromptTemplate.from_messages([
    ("system", '{prompt}'),
    ("human", '{content}')
])


def project(request: MyCustomRequest):
    return {
        "prompt": request.prompt,
        "content": request.content,
    }


async def process_response(input):
    async for chunk in input:
        yield chunk


chain = {
    'response': project | prompt | model,
    'original': RunnablePassthrough(),
} | RunnableGenerator(process_response)

[shaojun]

really looking forward for the context way, and it is very necessary for langserve applications.