Passing RunnableConfig through invoke, stream method in request body for a chain exposed through add_route method [langserve,langchain]

[tyadav7]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableConfig
from langserve import add_routes
from fastapi import FastAPI

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

import json

llm = ChatOpenAI()

def parse_or_fix(text: str, config: RunnableConfig):
    fixing_chain = (
        ChatPromptTemplate.from_template(
            "Fix the following text:\n\n\n{input}\n\nError: {error}"
            " Don't narrate, just respond with the fixed data."
        )
        | llm
        | StrOutputParser()
    )
    for _ in range(3):
        try:
            return json.loads(text)
        except Exception as e:
            text = fixing_chain.invoke({"input": text, "error": e}, config)
    return "Failed to parse"

app = FastAPI()
add_routes(app, RunnableLambda(parse_or_fix) , path='/parseorfixchain')

Description

I am trying to create a web server with langserve

I am exposing a chain as shown using add_routes method. I want to pass a runnable config and want to do the equivalent of

RunnableLambda(parse_or_fix).invoke("{foo: bar}", {"tags": ["my-tag"], "configurable": {"session_id":"" , "some_nore_config": ""}}

I want the route to show correct documentation on the docs page in langserve and also to pass this config inside my custom function.

System Info

System Information

OS: Windows
OS Version: 10.0.22621
Python Version: 3.10.0 (tags/v3.10.0:b494f59, Oct 4 2021, 19:00:18) [MSC v.1929 64 bit (AMD64)]

Package Information

langchain_core: 0.2.34
langchain: 0.2.14
langchain_community: 0.2.12
langsmith: 0.1.103
langchain_chroma: 0.1.3
langchain_cli: 0.0.30
langchain_openai: 0.1.22
langchain_text_splitters: 0.2.2
langserve: 0.2.2

Optional packages not installed

langgraph

[dosubot[bot]]

Hey there, @tyadav7! I'm here to help you with any bugs, questions, or contributions you might have. Let's tackle this together while we wait for a human maintainer.

To pass a RunnableConfig through the invoke method and ensure the correct documentation is shown on the docs page when exposing a chain using the add_routes method in langserve, you can follow the example provided below:

1. Define the RunnableConfig and pass it through the invoke method:

   from langchain_core.output_parsers import StrOutputParser
   from langchain_core.runnables import RunnableConfig
   from langserve import add_routes
   from fastapi import FastAPI
   
   from langchain_core.prompts import ChatPromptTemplate
   from langchain_core.runnables import RunnableLambda
   from langchain_openai import ChatOpenAI
   
   import json
   
   llm = ChatOpenAI()
   
   def parse_or_fix(text: str, config: RunnableConfig):
       fixing_chain = (
           ChatPromptTemplate.from_template(
               "Fix the following text:\n\n\n{input}\n\nError: {error}"
               " Don't narrate, just respond with the fixed data."
           )
           | llm
           | StrOutputParser()
       )
       for _ in range(3):
           try:
               return json.loads(text)
           except Exception as e:
               text = fixing_chain.invoke({"input": text, "error": e}, config)
       return "Failed to parse"
   
   app = FastAPI()
   add_routes(app, RunnableLambda(parse_or_fix), path='/parseorfixchain')

2. Invoke the chain with RunnableConfig:

   config = {
       "tags": ["my-tag"],
       "configurable": {"session_id": "", "some_more_config": ""}
   }
   result = RunnableLambda(parse_or_fix).invoke("{foo: bar}", config)

3. Ensure the correct documentation is shown on the docs page:

   • Follow standard documentation practices and ensure that the RunnableConfig usage is clearly documented in your code examples and API references.
   • Use the RunnableLambda constructor and handle run metadata and streaming with custom functions as shown in the example.

By following these steps, you can pass a RunnableConfig through the invoke method and ensure the correct documentation is shown on the docs page when exposing a chain using the add_routes method in langserve ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu No, I understand this but I want to pass this in the request body since I have added this route in langserve -> add_route

[dosubot[bot]]

To pass a RunnableConfig in the request body when using the add_route method in langserve, you can follow the example provided below. This example demonstrates how to define a FastAPI route that accepts a request body containing the RunnableConfig and then invokes the RunnableLambda with this configuration.

1. Define the FastAPI route and accept the request body:

   from fastapi import FastAPI, Request
   from pydantic import BaseModel
   from langchain_core.output_parsers import StrOutputParser
   from langchain_core.runnables import RunnableConfig, RunnableLambda
   from langchain_core.prompts import ChatPromptTemplate
   from langchain_openai import ChatOpenAI
   from langserve import add_routes
   import json
   
   llm = ChatOpenAI()
   
   class ConfigModel(BaseModel):
       tags: list[str]
       configurable: dict
   
   def parse_or_fix(text: str, config: RunnableConfig):
       fixing_chain = (
           ChatPromptTemplate.from_template(
               "Fix the following text:\n\n\n{input}\n\nError: {error}"
               " Don't narrate, just respond with the fixed data."
           )
           | llm
           | StrOutputParser()
       )
       for _ in range(3):
           try:
               return json.loads(text)
           except Exception as e:
               text = fixing_chain.invoke({"input": text, "error": e}, config)
       return "Failed to parse"
   
   app = FastAPI()
   
   @app.post("/parseorfixchain")
   async def parse_or_fix_route(request: Request):
       body = await request.json()
       text = body.get("text")
       config_data = body.get("config")
       config = RunnableConfig(**config_data)
       result = RunnableLambda(parse_or_fix).invoke(text, config)
       return {"result": result}
   
   add_routes(app, RunnableLambda(parse_or_fix), path='/parseorfixchain')

2. Example request body:

   {
       "text": "{foo: bar}",
       "config": {
           "tags": ["my-tag"],
           "configurable": {"session_id": "", "some_more_config": ""}
       }
   }

In this example, the FastAPI route /parseorfixchain accepts a JSON request body containing the text to be processed and the config for the RunnableConfig. The RunnableLambda is then invoked with these parameters.

This approach ensures that the RunnableConfig is passed through the request body and utilized within the parse_or_fix function.

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu Does the api itself not provide a way to pass the parameters
In the docs I can see the request body as

{
"input": "{foo:bar}",
"config": {"configurable": {"session_id": "", "some_more_config": ""}},
"kwargs": {}
}

passing config here does not work

[dosubot[bot]]

The API does provide a way to pass the parameters in the request body, but it must be done correctly according to the expected structure. The configuration should be passed under the "configurable" key within the "config" object, and the keys must match the expected keys for the function being invoked.

Here is the correct way to pass the configuration parameters:

{
  "input": "{foo:bar}",
  "config": {
    "configurable": {
      "session_id": "your_session_id",
      "some_more_config": "your_config_value"
    }
  },
  "kwargs": {}
}

Ensure that the keys in the "configurable" object match the expected keys for the function. For example, if you are using the RunnableWithMessageHistory class, the session_id is crucial for maintaining conversation history.

Here is an example of how to invoke a chain with the correct configuration:

chain_with_history.invoke(
    {"ability": "math", "question": "What does cosine mean?"},
    config={"configurable": {"session_id": "foo"}}
)

Or if the session factory takes multiple keys:

with_message_history.invoke(
    {"ability": "math", "question": "What does cosine mean?"},
    config={"configurable": {"user_id": "123", "conversation_id": "1"}}
)

Make sure that the keys in your request body match the expected keys for the function you are calling ^[1][2].

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu
I have sent the same payload that you gave me

{
"input": "{foo:bar}",
"config": {
"tags":["my-tag"],
"configurable": {
"session_id": "your_session_id",
"some_more_config": "your_config_value"
}
},
"kwargs": {}
}

here is a snippet of the data that I have received inside my parse_or_fix function
Probably its a bug and should be tracked

[dosubot[bot]]

To use the ensure_config function, you need to call it to ensure that your configuration dictionary is properly initialized and merged with default values. Here is how you can integrate the ensure_config function into your existing code:

1. Define the ensure_config function:
   Make sure you have the ensure_config function defined in your code. Here is the function with added debug statements for inspection:

   import logging
   from typing import Optional, cast
   from langchain_core.runnables import RunnableConfig
   
   logging.basicConfig(level=logging.DEBUG)
   
   def ensure_config(config: Optional[RunnableConfig] = None) -> RunnableConfig:
       logging.debug(f"Initial config: {config}")
       empty = RunnableConfig(
           tags=[],
           metadata={},
           callbacks=None,
           recursion_limit=25,  # Use the appropriate default value
           configurable={},
       )
       if var_config := var_child_runnable_config.get():
           empty.update(
               cast(
                   RunnableConfig,
                   {
                       k: v.copy() if k in COPIABLE_KEYS else v  # type: ignore[attr-defined]
                       for k, v in var_config.items()
                       if v is not None
                   },
               )
           )
       if config is not None:
           empty.update(
               cast(
                   RunnableConfig,
                   {
                       k: v.copy() if k in COPIABLE_KEYS else v  # type: ignore[attr-defined]
                       for k, v in config.items()
                       if v is not None and k in CONFIG_KEYS
                   },
               )
           )
       if config is not None:
           for k, v in config.items():
               if k not in CONFIG_KEYS and v is not None:
                   empty["configurable"][k] = v
       for key, value in empty.get("configurable", {}).items():
           if (
               not key.startswith("__")
               and isinstance(value, (str, int, float, bool))
               and key not in empty["metadata"]
           ):
               empty["metadata"][key] = value
       logging.debug(f"Final config: {empty}")
       return empty

2. Use the ensure_config function in your parse_or_fix function:
   Call the ensure_config function to ensure that the configuration is properly initialized and merged before using it.

   from fastapi import FastAPI, Request
   from pydantic import BaseModel
   from langchain_core.output_parsers import StrOutputParser
   from langchain_core.runnables import RunnableConfig, RunnableLambda
   from langchain_core.prompts import ChatPromptTemplate
   from langchain_openai import ChatOpenAI
   from langserve import add_routes
   import json
   
   llm = ChatOpenAI()
   
   class ConfigModel(BaseModel):
       tags: list[str]
       configurable: dict
   
   def parse_or_fix(text: str, config: RunnableConfig):
       config = ensure_config(config)  # Ensure the config is properly initialized
       fixing_chain = (
           ChatPromptTemplate.from_template(
               "Fix the following text:\n\n\n{input}\n\nError: {error}"
               " Don't narrate, just respond with the fixed data."
           )
           | llm
           | StrOutputParser()
       )
       for _ in range(3):
           try:
               return json.loads(text)
           except Exception as e:
               text = fixing_chain.invoke({"input": text, "error": e}, config)
       return "Failed to parse"
   
   app = FastAPI()
   
   @app.post("/parseorfixchain")
   async def parse_or_fix_route(request: Request):
       body = await request.json()
       text = body.get("input")
       config_data = body.get("config")
       config = RunnableConfig(**config_data)
       result = RunnableLambda(parse_or_fix).invoke(text, config)
       return {"result": result}
   
   add_routes(app, RunnableLambda(parse_or_fix), path='/parseorfixchain')

3. Test the API:
   Send a request to the /parseorfixchain endpoint with the following JSON body:

   {
     "input": "{foo:bar}",
     "config": {
       "tags": ["my-tag"],
       "configurable": {
         "session_id": "your_session_id",
         "some_more_config": "your_config_value"
       }
     },
     "kwargs": {}
   }

By using the ensure_config function, you ensure that the configuration is correctly initialized and merged with default values before being used in your parse_or_fix function. This should help in correctly populating the configurable dictionary and other configuration parameters.

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu I want to pass the configurable parameters at runtime. And i have ensured that the configurable parameters are not added. Hence this is probably a bug or maybe both of us are looking at this wrong.

I think one of the maintainers might have an answer.

[dosubot[bot]]

A maintainer has already been looped in.

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu
After a lot of debugging I have finally figure out a way to work with it

So the add_routes takes an additional parameter called CONFIG_KEYS when passed it maps the body to the schema provided.

from langchain_core.runnables.config import COPIABLE_KEYS
add_routes(router, RunnableLambda(myfunc), config_keys=COPIABLE_KEYS , path='/agent')

But this still wont map the additional configurable fields needed to be set for the runnable function.

So I created a separate class which extends the RunnableLambda and overrides the method config_specs this override is needed so that the configurable parameters could be map so that FastAPI could then resolve this schema later when this lambda is called.

from typing import List, Sequence, Callable, Type, Union, Awaitable, Optional, Iterator
from langchain_core.runnables import (
    RunnableConfig,
    RunnableLambda,
    ConfigurableFieldSpec,
)
from langchain_core.runnables.utils import get_unique_config_specs
from typing import AsyncIterator
from langchain_core.runnables.utils import get_unique_config_specs, Input, Output
from langchain_core.callbacks.manager import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
)


class ModifiedRunnableLambda(RunnableLambda):

    def __init__(
        self,
        func: Union[
            Union[
                Callable[[Input], Output],
                Callable[[Input], Iterator[Output]],
                Callable[[Input, RunnableConfig], Output],
                Callable[[Input, CallbackManagerForChainRun], Output],
                Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output],
            ],
            Union[
                Callable[[Input], Awaitable[Output]],
                Callable[[Input], AsyncIterator[Output]],
                Callable[[Input, RunnableConfig], Awaitable[Output]],
                Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
                Callable[
                    [Input, AsyncCallbackManagerForChainRun, RunnableConfig],
                    Awaitable[Output],
                ],
            ],
        ],
        afunc: Optional[
            Union[
                Callable[[Input], Awaitable[Output]],
                Callable[[Input], AsyncIterator[Output]],
                Callable[[Input, RunnableConfig], Awaitable[Output]],
                Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
                Callable[
                    [Input, AsyncCallbackManagerForChainRun, RunnableConfig],
                    Awaitable[Output],
                ],
            ]
        ] = None,
        name: Optional[str] = None,
        factory_config: Sequence[ConfigurableFieldSpec] = [],
    ) -> None:
        """Create a RunnableLambda from a callable, and async callable or both.

        Accepts both sync and async variants to allow providing efficient
        implementations for sync and async execution.

        Args:
            func: Either sync or async callable
            afunc: An async callable that takes an input and returns an output.
                Defaults to None.
            name: The name of the Runnable. Defaults to None.
            factory_config: Configure fields that should be passed to the
            chat history factory. See ``ConfigurableFieldSpec`` for more details.

        Raises:
            TypeError: If the func is not a callable type.
            TypeError: If both func and afunc are provided.
        """
        super().__init__(func=func, afunc=afunc, name=name)
        self.factory_config = factory_config
        

    @property
    def config_specs(self) -> List[ConfigurableFieldSpec]:
        """Get the configuration specs for the A1RunnableLambda"""
        return get_unique_config_specs(super().config_specs + self.factory_config)

the new implementation of this will be

custom_config_specs = [
    ConfigurableFieldSpec(
        id="user_id",
        annotation=str,
        name="User ID",
        description="Unique identifier for the user.",
        default="",
        is_shared=True,
    ),
    ConfigurableFieldSpec(
        id="conversation_id",
        annotation=str,
        name="Conversation ID",
        description="Unique identifier for the conversation.",
        default="",
        is_shared=True,
    ),
]
add_routes(router, ModifiedRunnableLambda(myfunc, factory_config=custom_config_specs ), config_keys=COPIABLE_KEYS , path='/agent')

@dosu However this should be handled by the API itself if the payload consists of the configurable keys this should be passed to myfunc which is not handled atm, also my implementation shows that configurable keys are known before hand which is not an acceptable way of dealing with configurable IMHO, this should be raised as a bug and handled in some subsequent release.

[dosubot[bot]]

Great job on figuring out the solution! Extending RunnableLambda and overriding the config_specs method to include the additional configurable parameters is a smart approach. This ensures that FastAPI can resolve the schema correctly when the lambda is called.

Here's a summary of your solution for clarity:

1. Extend RunnableLambda:
   Create a new class ModifiedRunnableLambda that extends RunnableLambda and overrides the config_specs method to include the additional configurable parameters.

   from typing import List, Sequence, Callable, Type, Union, Awaitable, Optional, Iterator
   from langchain_core.runnables import (
       RunnableConfig,
       RunnableLambda,
       ConfigurableFieldSpec,
   )
   from langchain_core.runnables.utils import get_unique_config_specs
   from typing import AsyncIterator
   from langchain_core.callbacks.manager import (
       AsyncCallbackManagerForChainRun,
       CallbackManagerForChainRun,
   )
   
   class ModifiedRunnableLambda(RunnableLambda):
   
       def __init__(
           self,
           func: Union[
               Union[
                   Callable[[Input], Output],
                   Callable[[Input], Iterator[Output]],
                   Callable[[Input, RunnableConfig], Output],
                   Callable[[Input, CallbackManagerForChainRun], Output],
                   Callable[[Input, CallbackManagerForChainRun, RunnableConfig], Output],
               ],
               Union[
                   Callable[[Input], Awaitable[Output]],
                   Callable[[Input], AsyncIterator[Output]],
                   Callable[[Input, RunnableConfig], Awaitable[Output]],
                   Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
                   Callable[
                       [Input, AsyncCallbackManagerForChainRun, RunnableConfig],
                       Awaitable[Output],
                   ],
               ],
           ],
           afunc: Optional[
               Union[
                   Callable[[Input], Awaitable[Output]],
                   Callable[[Input], AsyncIterator[Output]],
                   Callable[[Input, RunnableConfig], Awaitable[Output]],
                   Callable[[Input, AsyncCallbackManagerForChainRun], Awaitable[Output]],
                   Callable[
                       [Input, AsyncCallbackManagerForChainRun, RunnableConfig],
                       Awaitable[Output],
                   ],
               ]
           ] = None,
           name: Optional[str] = None,
           factory_config: Sequence[ConfigurableFieldSpec] = [],
       ) -> None:
           super().__init__(func=func, afunc=afunc, name=name)
           self.factory_config = factory_config
   
       @property
       def config_specs(self) -> List[ConfigurableFieldSpec]:
           return get_unique_config_specs(super().config_specs + self.factory_config)

2. Use the ModifiedRunnableLambda in add_routes:
   Pass the ModifiedRunnableLambda to add_routes along with the config_keys parameter.

   from langchain_core.runnables.config import COPIABLE_KEYS
   add_routes(router, ModifiedRunnableLambda(myfunc), config_keys=COPIABLE_KEYS, path='/agent')

This approach ensures that the configurable parameters are correctly mapped and passed at runtime, allowing FastAPI to resolve the schema properly. Well done on debugging and finding this solution!

To continue talking to Dosu, mention @dosu.

[tyadav7]

@dosu This should be marked as a bug though because of the other details I mentioned earlier.

[dosubot[bot]]

A maintainer has already been looped in.

To continue talking to Dosu, mention @dosu.