Langchain动态路由绑定异常：非Langchain问题触发TypeError

错误原因 #

当路由到RAG检索链时，传入的是包含topic和question的字典对象，但retrieval_chain中使用RunnablePassthrough()直接将整个字典赋值给question参数，导致向量检索时把字典当作查询字符串传入embedding函数，触发TypeError: expected string or buffer（embedding函数需要字符串输入，而非字典）。

修复步骤 #

1. 修正RAG链的参数映射 #

将retrieval_chain中的question映射改为从输入字典中提取question字段，而非传递整个输入对象：

retrieval_chain = (
    {"context": retriever, "question": lambda x: x["question"]}  # 这里修改为提取question字段
    | prompt
    | model
    | StrOutputParser()
)

2. 删除冗余错误代码 #

移除代码中无效的片段：

# 删掉以下错误代码
response = full
    | StrOutputParser()
)

3. 优化变量初始化（可选） #

将retrieval_chain的初始值从空字符串改为None，避免类型混淆：

retrieval_chain = None

修复后完整代码 #

classification_chain = (
    PromptTemplate.from_template(
        """Given the user question below, classify it as either being about 'Langchain' or 'something else'\
        Do not respond with more than one word.

        <question>
        {question}
        </question>

Classification:"""
    )
    | get_llm()
    | StrOutputParser()
)


langchain_chain = (
    PromptTemplate.from_template(
        """You are an expert in langchain. Always answer with Daddy says.\
            <question>
            {question}
            </question>
            Answer:"""
    )
    | get_llm()
    | StrOutputParser()
)


retrieval_chain = None 

def route(info):
    if "langchain" in info["topic"].lower():
        return langchain_chain
    else:
        print('-'*100)
        print(retrieval_chain)
        return retrieval_chain


def get_RAG_response(collection_name: str, question: str):
    db = get_vector_db(collection_name)
    retriever = db.as_retriever()
    model = get_llm()
    template = """Answer the following question based only on the provided context:
    {context}
    
    Question: {question}
    """
    prompt = ChatPromptTemplate.from_template(template)
    global retrieval_chain
    retrieval_chain = (
        {"context": retriever, "question": lambda x: x["question"]}
        | prompt
        | model
        | StrOutputParser()
    )

    full_chain = {"topic": classification_chain, "question": lambda x: x["question"]} | RunnableLambda(route)
    response = full_chain.invoke({"question": question})
    print('*'*100)
    print('response:', response)
    return response

内容的提问来源于stack exchange，提问作者garg10may