LangChain RunnableParallel — dict output shape not matching downstream

from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

# Prompt uses {context} and {question}
prompt = ChatPromptTemplate.from_template(
    "Given this context:\n{context}\n\nAnswer: {question}"
)

def retrieve(question: str) -> str:
    # Your vector store or search
    return "Relevant docs about " + question

# ❌ WRONG — branch names do not match template variables
# parallel_wrong = RunnableParallel(docs=retrieve, q=RunnablePassthrough())
# chain = parallel_wrong | prompt | model    # KeyError: 'context'

# ✓ RIGHT — branch names match {context} and {question}
parallel = RunnableParallel(
    context=itemgetter("question") | RunnableLambda(retrieve),
    question=itemgetter("question"),   # or RunnablePassthrough() if input is str
)

chain = parallel | prompt | model | StrOutputParser()
print(chain.invoke({"question": "What is LCEL?"}))

# ✓ CLEANER — RunnablePassthrough.assign() adds context, keeps question
chain2 = (
    RunnablePassthrough.assign(
        context=lambda x: retrieve(x["question"])
    )
    | prompt
    | model
    | StrOutputParser()
)
print(chain2.invoke({"question": "What is LCEL?"}))

from operator import itemgetter
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

# Fan out to two independent processes
parallel = RunnableParallel(
    summary=summarise_chain,
    keywords=keywords_chain,
)

# Downstream we only need the summary — extract with itemgetter
summary_only_chain = parallel | itemgetter("summary")

# For downstream that needs both, keep the dict shape
both_chain = parallel | some_step_that_takes_dict

# Combining multiple itemgetters
def render(x):
    return f"Summary: {x['summary']}\n\nKeywords: {', '.join(x['keywords'])}"

rendered_chain = parallel | render

# When you need to reshape a dict:
reshape_chain = (
    parallel
    | (lambda x: {"final_summary": x["summary"], "final_tags": x["keywords"]})
)

# For very long-form transformations, use RunnableLambda for tracing benefits
from langchain_core.runnables import RunnableLambda
labeled_chain = parallel | RunnableLambda(
    lambda x: {"final_summary": x["summary"], "final_tags": x["keywords"]},
    name="reshape_output",
)

from langchain_core.runnables import RunnableParallel, RunnableLambda

# Flaky retrieval — wrap it so failures return an empty result instead of throwing
def safe_retrieve(question: str) -> str:
    try:
        return retrieve(question)
    except Exception as e:
        # Log to your observability stack
        print(f"[retrieve failed] {e}")
        return ""  # graceful degradation — chain continues with empty context

# Per-branch fallback using with_fallbacks
from langchain_core.runnables import RunnableLambda

primary_retrieve = RunnableLambda(retrieve)
fallback_retrieve = RunnableLambda(lambda q: "")

resilient_retrieve = primary_retrieve.with_fallbacks([fallback_retrieve])

parallel = RunnableParallel(
    context=itemgetter("question") | resilient_retrieve,
    question=itemgetter("question"),
)

# If retrieve fails, context becomes "" and the chain still runs
chain = parallel | prompt | model | StrOutputParser()

# For async chains, be aware: one branch raising will cancel the others
# unless you set return_exceptions=True in the underlying gather.
# The simplest defense is to make each branch self-recovering as above.