How do I create a function that takes the input_type, Configurable fields, and abitrary input?

[joshuajohananorthos]

I keep running into an issue where I want to take the input, configuration for the models (api_key, endpoint, etc) and a Runnable Config?

I need an endpoint to give me three things:

• The input type
• Configurable fields bound with the chain
• Configuration settings that are stored in the Request

I know that I can use per_req_config_modifier, but that will only bind keys that are defined as fields on the chain or fields in custom types that I create beforehand. I cannot figure out how to bind both the chain and a custom type at the same time.

Here is an example of the custom type I created:

class ConfigurableLambda(RunnableSerializable, RunnableLambda):
    afunc: Any

    def __init__(self, **data):
        super().__init__(**data)
        # get afunc from data
        afunc = data.get("afunc")
        RunnableLambda.__init__(self, afunc)

# new class with fields that can be bound
class WithApiKey(ConfigurableLambda):
    open_api_key: Optional[str] = Field(None, description="This is the api key")
    azure_endpoint: Optional[str] = Field(None, description="This is the endpoint")

# implement the class
tester = WithApiKey(afunc=test_async, input_type=NamespaceRequest).configurable_fields(
    open_api_key=ConfigurableField(id="openai_api_key"),
    azure_endpoint=ConfigurableField(id="azure_endpoint"),
    )

# function to execute the models
def test(req: NamespaceRequest, config: RunnableConfig):
    index = pinecone.Index("test")

    openai_embeddings = AzureOpenAIEmbeddings(
        api_version="2023-07-01-preview",
        azure_endpoint=config["configurable"]["azure_endpoint"],
        azure_deployment="embed_deploy",
        api_key=config["configurable"]["openai_api_key"],
    )
        
    vectorstore = Pinecone(
        index, openai_embeddings, "text", namespace=req.namespace
    )

    aco = AzureChatOpenAI(
        api_version="2023-07-01-preview",
        azure_endpoint=config["configurable"]["azure_endpoint"],
        azure_deployment="gpt_deploy",
        api_key=config["configurable"]["openai_api_key"],
        temperature=0.0
    )

    qa_with_sources = RetrievalQAWithSourcesChain.from_chain_type(
        llm=aco,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(),
        return_source_documents=True
    )
    # I cannot add configurable_alternatives here as they will never be seen

    prompt = PromptTemplate.from_template("{prompt}: {question}")
    chain = (
        RunnableLambda(lambda x: { "question": x.question, "prompt": x.prompt })
        | prompt
        | RunnableLambda(lambda x: { "question": x.text })
        | qa_with_sources
    )

    return chain

# without an async def the Lambda will destroy the input arguments
async def test_async(req: NamespaceRequest, config: RunnableConfig):
    return await asyncio.get_running_loop().run_in_executor(None, test, req, config)

# per req
def get_app(config: Dict[str, Any], req: Request):
    config["configurable"]["openai_api_key"] = req.app.state.openai_api_key
    config["configurable"]["azure_endpoint"] = req.app.state.azure_endpoint
    return config

# add routes
add_routes(router, tester, per_req_config_modifier=get_app, path="/prompt_tester")

This works, but I cannot then add other configurable fields into the Chain. Technically I could add them to the outer type and pass them down or add logic to determine what to do.

This is a workaround of the execution for why Configurable alternatives were created.

I cannot just create the chain and use that in add_routes as it needs per request information. Then if I create a function to do all of that, then I lose the context of the chain as add_routes inspects the function not the returned chain. Which is what lead to the types to establish a context around the function.

I would like to create a parameterized function that asks for everything it needs to execute the chain without having to create all the objects in the root of a FastApi project.

It would great if I could create a chain that would execute like this:

@router.post("/example")
async def example(embedding : AzureOpenAIEmbeddings = Depends(get_embedding)):
    # embedding object is injected here with everything it needs

[eyurtsev]

@joshuajohananorthos Runnable Lambdas are not configurable right now so what you did works. Ideal end state is obviously closer to

@router.post("/example")
async def example(embedding : AzureOpenAIEmbeddings = Depends(get_embedding)):
    # embedding object is injected here with everything it needs

I'm trying to expose internals now to make this easier.

By the way, runnable lambdas don't support streaming, so if using runnable lambdas instead of runnable generators double check that streaming still works / or that it's not needed.

[joshuajohananorthos]

@eyurtsev Thank you for the response.

Can you give me a little more insight on how I would make the chain configurable, for example changing the llm? In my example the configurations don't get passed to the chain.

Would I have to add those options to the WithApiKey class?

How would you implement this? That is passing configuration to the init methods and passing configurations to the chain.

[eyurtsev]

Ah sorry I thought that you got that working as a workaround. I'll try to take a look at how to do this. You'll probably need to inherit from RunnableSerializer, create properties that you want to be considered as dynamic fields, and and then delegate invoke to the underlying RunnableLambda.

Without looking more deeply not sure that you'll just be able to inherit from both and get it working.

[eyurtsev]

@joshuajohananorthos does something like this work:

langserve/examples/api_handler_simple_server/server.py

Line 96 in 69a33bc

request: Request, runnable: Annotated[APIHandler, Depends(_get_api_handler)]

This is a proposal in this PR: #333

[eyurtsev]

I think it requires one more tweak to allow you to pass in additional config payload as a dict that's specified by the server