How to implement conditional retriever selection in LangChain workflows

[CreativeArtist88]

I’m working on a system where users pick a service type and ask questions. Based on their choice, I need to query different Elasticsearch databases and generate answers.

I have three services, each with its own search index. My current approach creates separate chains:

service_x_chain = search_x | template | model
service_y_chain = search_y | template | model  
service_z_chain = search_z | template | model

This works but I want to build one unified chain that picks the right searcher automatically:

searchers = {
    "service_x": search_x,
    "service_y": search_y,
    "service_z": search_z
}

def pick_searcher(data):
    return searchers[data['service_type']]

unified_chain = RunnableLambda(pick_searcher) | template | model

result = unified_chain.invoke({
    "user_query": "What causes ocean waves",
    "service_type": "service_y"
})

The issue is that my router function correctly identifies which searcher to use, but then it tries to run that searcher with the full dictionary input. LangChain retrievers only accept plain strings as queries, not dictionaries with multiple keys.

I need to route to the correct retriever while also extracting just the query string for the searcher. Any suggestions on how to handle this parameter passing issue?

[ryanl]

Your router function picks the searcher but doesn’t format the input for what comes next. I’ve hit this same issue building multi-tenant search systems.

Combine both operations in your lambda:

def route_and_search(data):
    searcher = searchers[data['service_type']]
    return searcher.invoke(data['user_query'])

unified_chain = RunnableLambda(route_and_search) | template | model

Now your lambda picks the right searcher AND calls it with the correct string parameter.

If you want more control, try RunnablePassthrough with itemgetter:

from operator import itemgetter

def conditional_search(data):
    query = data['user_query']
    service = data['service_type'] 
    results = searchers[service].invoke(query)
    return {"context": results, "query": query}

unified_chain = RunnablePassthrough() | RunnableLambda(conditional_search) | template | model

I’d go with the first option - it’s cleaner. Your routing function should handle the complete transformation, not just selection.

This video covers some solid patterns for connecting external data sources like Elasticsearch with LLM workflows:

ChatGPT and Elasticsearch: OpenAI meets private data