LangChain中SQLDatabaseChain结合MultiPromptChain报错求助

原本使用单个SQLDatabaseChain，因默认提示词包含表格/普通文本两种输出格式逻辑，内容过长导致消耗过多OpenAI额度。计划拆分出两个带独立提示词的SQLDatabaseChain，通过MultiPromptChain路由调用，但运行时出现ValidationError。

from sqlalchemy import *
from sqlalchemy.engine import create_engine
from sqlalchemy.schema import *
import os
from flask import jsonify, Flask, make_response

from langchain.prompts.prompt import PromptTemplate
from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain import SQLDatabase, SQLDatabaseChain
from langchain.llms.openai import OpenAI
from langchain.agents import AgentExecutor
from langchain.callbacks import get_openai_callback
import json
import re


def main(query):
    global db_chain

    prompt_default = ("""

You are a Sales Bot, and your goal is to provide answers based on the sales data stored in the database.

To retrieve information from the database, follow this process:

Receive a question or query from the user.
Formulate a syntactically correct query based on the question. Make sure to include relevant table names, columns, conditions, and any necessary aggregations or joins.
Execute the query on the sales database.
Examine the results obtained from the query.
Provide the answer to the user in the required format.
If answer is not required in table format, here's the format to use when responding:

Question: "Question from the user"
Query: "Query formulated based on the question"
Result: "Results obtained from executing the query"
Answer: "Final answer based on the result"

If answer is required in table format, respond in the following format:

{"table": {"columns": ["column1", "column2", ...], "data": [["value1", "value2", ...], ["value1", "value2", ...], ...]}}

Ensure that all strings in the "columns" list and data list are enclosed in double quotes.

For example:
{"table": {"columns": ["product_name", "sales_quantity"], "data": [["Product A", 100], ["Product B", 200], ...]}}

If you are unable to provide an answer, respond as follows:

{"answer": "I'm sorry, but I don't have that information at the moment."}

Remember to return all output as a string.

Now, let's proceed step by step. Below is the query:

Query: 


""" + query )

    sqlalchemy_url = f'sqlite:///../../../../notebooks/Chinook.db'

    # OpenAI API key
    os.environ["OPENAI_API_KEY"] = ""

    # Set up langchain
    db = SQLDatabase.from_uri(sqlalchemy_url, view_support=True)
    llm = OpenAI(temperature=0, model="text-davinci-003",max_tokens=1000)
    db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, use_query_checker=True, top_k=20)

    # First query
    result = db_chain.run(prompt_default)
    return result

query_run = main("Give me top 5 stock codes in a table format")

print(query_run)

import os
from langchain.chains.router import MultiPromptChain
from langchain.llms import OpenAI
from langchain.chains import ConversationChain, SQLDatabaseSequentialChain
from langchain.chains.llm import LLMChain
from langchain.prompts import PromptTemplate
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
from langchain import OpenAI, SQLDatabase, SQLDatabaseChain

from sqlalchemy import *
from sqlalchemy.engine import create_engine
from sqlalchemy.schema import *
import os
from flask import jsonify, Flask, make_response

table_template = """
You are a Sales Bot, and your goal is to provide answers based on the sales data stored in the BigQuery database.

BigQuery use variant of SQL that is standard SQL so your query should be standard sql query.

To retrieve information from the database, follow this process:

Receive a question or query from the user.
Formulate a syntactically correct BigQuery query based on the question. Make sure to include relevant table names, columns, conditions, and any necessary aggregations or joins.
Execute the BigQuery query on the sales database.
Examine the results obtained from the query.
Provide the answer to the user in the following format:

{"table": {"columns": ["column1", "column2", ...], "data": [["value1", "value2", ...], ["value1", "value2", ...], ...]}}

Ensure that all strings in the "columns" list and data list are enclosed in double quotes.

For example:
{"table": {"columns": ["product_name", "sales_quantity"], "data": [["Product A", "100"], ["Product B", "200"], ...]}}

If you are unable to provide an answer, please reply "I'm sorry, but I don't have that information at the moment.".

Remember to return all output as a string.

Now, let's proceed step by step. Below is the query:

Query:

{input}
"""


ans_template = """
You are a Sales Bot, and your goal is to provide answers based on the sales data stored in the BigQuery database.

BigQuery use variant of SQL that is standard SQL so your query should be standard sql query.

To retrieve information from the database, follow this process:

Receive a question or query from the user.
Formulate a syntactically correct BigQuery query based on the question. Make sure to include relevant table names, columns, conditions, and any necessary aggregations or joins.
Execute the BigQuery query on the sales database.
Examine the results obtained from the query.
Provide the answer to the user in the required format.
If answer is not required in table format, here's the format to use when responding:

Question: "Question from the user"
BigQuery Query: "BigQuery query formulated based on the question"
BigQuery Result: "Results obtained from executing the BigQuery query"
Answer: "Final answer based on the BigQuery result"

If you are unable to provide an answer, please reply as "I'm sorry, but I don't have that information at the moment.".

Remember to return all output as a string.

Now, let's proceed step by step. Below is the query:

Query:

{input}
"""
prompt_infos = [
    {
        "name": "table_format",
        "description": "Good for answering questions if user asks to generate a table",
        "prompt_template": table_template
    },
    {
        "name": "ans_format",
        "description": "Good for answering questions if user don't asks for any specific format",
        "prompt_template": ans_template
    }
]

os.environ["OPENAI_API_KEY"] = ""

llm = OpenAI(temperature=0, model="text-davinci-003", max_tokens=1000)
sqlalchemy_url = f'sqlite:///../../../../notebooks/Chinook.db'
db = SQLDatabase.from_uri(sqlalchemy_url, view_support=True)

destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt_template = p_info["prompt_template"]
    prompt = PromptTemplate(template=prompt_template, input_variables=["input"], validate_template=False)
    chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, prompt=prompt, use_query_checker=True, top_k=20)
    destination_chains[name] = chain
default_chain = ConversationChain(llm=llm, output_key="text")

destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=destinations_str
)
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
    validate_template=False
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

chain = MultiPromptChain(router_chain=router_chain, destination_chains=destination_chains, default_chain=default_chain, verbose=True)

print(chain.run("Give me top 5 stock codes in a table format"))

ValidationError: 20 validation errors for MultiPromptChain
destination_chains -> table_format -> prompt
  none is not an allowed value (type=type_error.none.not_allowed)
destination_chains -> table_format -> llm
  none is not an allowed value (type=type_error.none.not_allowed)
destination_chains -> table_format -> database
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> input_key
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> llm_chain
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> query_checker_prompt
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> return_direct
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> return_intermediate_steps
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> top_k
  extra fields not permitted (type=value_error.extra)
destination_chains -> table_format -> use_query_checker
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> prompt
  none is not an allowed value (type=type_error.none.not_allowed)
destination_chains -> ans_format -> llm
  none is not an allowed value (type=type_error.none.not_allowed)
destination_chains -> ans_format -> database
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> input_key
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> llm_chain
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> query_checker_prompt
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> return_direct
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> return_intermediate_steps
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> top_k
  extra fields not permitted (type=value_error.extra)
destination_chains -> ans_format -> use_query_checker
  extra fields not permitted (type=value_error.extra)

问题核心是MultiPromptChain要求目标链为LLMChain类型，但传入的SQLDatabaseChain结构不兼容，导致字段验证失败。需要将SQLDatabaseChain包装成符合要求的链结构。

修改后的完整代码 #

import os
from langchain.chains.router import MultiPromptChain
from langchain.llms import OpenAI
from langchain.chains import ConversationChain
from langchain.chains.base import Chain
from langchain.prompts import PromptTemplate
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
from langchain import SQLDatabase, SQLDatabaseChain

table_template = """
You are a Sales Bot, and your goal is to provide answers based on the sales data stored in the database.

Formulate a syntactically correct SQL query based on the question, execute it, and return results in the following table format:

{"table": {"columns": ["column1", "column2", ...], "data": [["value1", "value2", ...], ["value1", "value2", ...], ...]}}

Ensure all strings in columns and data are enclosed in double quotes. If unable to answer, reply "I'm sorry, but I don't have that information at the moment.".

Query: {input}
"""

ans_template = """
You are a Sales Bot, and your goal is to provide answers based on the sales data stored in the database.

Formulate a syntactically correct SQL query based on the question, execute it, and return results in this format:

Question: "{input}"
Query: "Generated SQL query"
Result: "Query results"
Answer: "Final answer based on result"

If unable to answer, reply "I'm sorry, but I don't have that information at the moment.".
"""

prompt_infos = [
    {
        "name": "table_format",
        "description": "Good when user explicitly asks for a table format",
        "prompt_template": table_template
    },
    {
        "name": "ans_format",
        "description": "Good when user doesn't specify any output format",
        "prompt_template": ans_template
    }
]

os.environ["OPENAI_API_KEY"] = ""

llm = OpenAI(temperature=0, model="text-davinci-003", max_tokens=1000)
sqlalchemy_url = f'sqlite:///../../../../notebooks/Chinook.db'
db = SQLDatabase.from_uri(sqlalchemy_url, view_support=True)

# 自定义包装类，将SQLDatabaseChain适配为Chain抽象类要求的结构
class SQLChainWrapper(Chain):
    db_chain: SQLDatabaseChain
    input_key: str = "input"
    output_key: str = "text"
    
    @property
    def input_keys(self):
        return [self.input_key]
    
    @property
    def output_keys(self):
        return [self.output_key]
    
    def _call(self, inputs):
        result = self.db_chain.run(inputs[self.input_key])
        return {self.output_key: result}

destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt_template = p_info["prompt_template"]
    # 创建带定制prompt的SQLDatabaseChain
    db_chain = SQLDatabaseChain.from_llm(llm, db, verbose=True, 
                                        prompt=PromptTemplate(template=prompt_template, input_variables=["input"]),
                                        use_query_checker=True, top_k=20)
    # 包装成符合MultiPromptChain要求的链
    wrapped_chain = SQLChainWrapper(db_chain=db_chain)
    destination_chains[name] = wrapped_chain

default_chain = ConversationChain(llm=llm, output_key="text")

# 构建路由链
destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations=destinations_str)
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
    validate_template=False
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

# 初始化MultiPromptChain
chain = MultiPromptChain(router_chain=router_chain, destination_chains=destination_chains, default_chain=default_chain, verbose=True)

print(chain.run("Give me top 5 stock codes in a table format"))

关键修改点 #

1. 新增SQLChainWrapper类，将SQLDatabaseChain包装成符合LangChain抽象Chain类的结构，明确输入输出键，确保返回{"text": 结果}的字典格式，匹配MultiPromptChain的预期。
2. 替换原直接将SQLDatabaseChain存入destination_chains的逻辑，改用包装后的链实例，解决结构不兼容导致的字段验证错误。

内容的提问来源于stack exchange，提问作者Kamran Khan Khilji