LangChain链式调用Runnables触发TypeError的问题求助

问题描述 #

在用LangChain构建基于检索的QA系统时，链式调用Runnables遇到TypeError，错误出现在使用|管道操作符连接RunnablePassthrough、自定义prompt和ChatOpenAI实例的环节。

错误信息 #

TypeError: Expected a Runnable, callable or dict. Instead got an unsupported type: <class 'str'>

问题代码段 #

错误定位至以下代码：

rag_chain = ( {"context": context, "question": RunnablePassthrough()} | rag_custom_prompt | llm )

预期RunnablePassthrough()能将context和question传递到链的下一步，但转换为Runnable时失败。

完整代码 #

## Convert the pdf into txt
def pdf_to_txt(inst_manuals):

    txt = ""
    for manual in inst_manuals:
        reader = PdfReader(inst_manuals)
        for page in reader.pages:
            txt += page.extract_text()

    return txt

## Convert txt into chunks 
def chunkify_txt(txt):

    txt_splitter = CharacterTextSplitter(
        separator= "\n",
        chunk_size= 1000,
        chunk_overlap= 200,
        length_function= len
    )

    chunks = txt_splitter.split_text(txt)

    return chunks

## Obtain the vector store
def get_vector(chunks):
    embeddings = OpenAIEmbeddings()

    vectorstore = FAISS.from_texts(texts= chunks, embedding = embeddings)

    return vectorstore

## Retrieve useful info similar to user query
def retrieve(vectorstore, question):
    logging.basicConfig()
    logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

    retriever_from_llm = MultiQueryRetriever.from_llm(
        retriever=vectorstore.as_retriever(), llm=ChatOpenAI(temperature=0)
    )
    unique_docs = retriever_from_llm.get_relevant_documents(query=question)
    
    print(f"Number of unique documents retrieved: {len(unique_docs)}")
    
    return unique_docs
    

## Generate response for user query

def gen_resp(retriever, question):
    llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
    template = """... [custom prompt template] ..."""
    rag_custom_prompt = PromptTemplate.from_template(template)

    context = "\n".join(doc.page_content for doc in retriever)

    rag_chain = (
        {"context": context, "question": RunnablePassthrough()} | rag_custom_prompt | llm
    )

    answer = rag_chain.invoke(question)

    return answer

已尝试方法 #

• 使用不同检索器
• 查阅LangChain官方文档中Runnables及链式操作的用法
• 交换rag_chain字典中context和question的键，排查顺序问题

问题原因及解决方案 #

核心问题 #

链起始字典中直接传入了字符串类型的context，但LangChain的Runnable管道要求字典内每个值必须是Runnable、可调用对象或字典，普通字符串不符合输入要求，因此触发TypeError。

修复方案 #

方案一：用RunnableLambda包装context

将字符串类型的context包装为返回该字符串的RunnableLambda，确保字典内所有值均为Runnable类型：

from langchain_core.runnables import RunnableLambda

# 修改rag_chain定义
rag_chain = (
    {"context": RunnableLambda(lambda _: context), "question": RunnablePassthrough()} 
    | rag_custom_prompt 
    | llm
)

RunnableLambda(lambda _: context)会忽略输入参数，直接返回预定义的context字符串，适配管道要求。

方案二：直接将context注入prompt模板

若context在调用链前已确定，可直接将其填入prompt模板，无需在Runnable字典中传递：

# 直接填充prompt模板
filled_prompt = rag_custom_prompt.format(context=context, question=question)
# 调用LLM生成回答
answer = llm.invoke(filled_prompt)

该方式更简洁，适合context提前确定的场景。

额外优化建议 #

1. 修复pdf_to_txt函数BUG：当前函数中reader = PdfReader(inst_manuals)应改为reader = PdfReader(manual)，否则循环时会重复读取整个文件列表：

def pdf_to_txt(inst_manuals):
    txt = ""
    for manual in inst_manuals:
        reader = PdfReader(manual)  # 修正为manual，而非inst_manuals
        for page in reader.pages:
            txt += page.extract_text()
    return txt

2. 避免重复创建LLM实例：retrieve和gen_resp函数均创建了ChatOpenAI实例，可将LLM实例作为参数传入或全局定义，减少资源消耗。

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