问题在于，GraphCypherQAChain类有一个字段用于存储input_key,默认值为"query"。

input_key: str = "query"

然后这个字段被用来检索问题。请参阅GraphCypherQAChain类中_call方法的代码：

question = inputs[self.input_key]

        intermediate_steps: List = []

        generated_cypher = self.cypher_generation_chain.run(
            {"question": question, "schema": self.graph_schema}, callbacks=callbacks
        )

如果你将代码更改为：

res = chain.invoke({"schema": graph.schema, "examples": examples, "query": question})

那么你的代码就可以正常工作了。

感谢@RafaelXokito。问题仍然存在。
代码

res = chain.invoke({"schema": graph.schema, "examples": examples, "query": question})

错误(与上述类似)

...
  File "/Users/pierreoberholzer/code/knowledge_graphs/venv/lib/python3.11/site-packages/langchain/chains/base.py", line 284, in _validate_inputs
    raise ValueError(f"Missing some input keys: {missing_keys}")
ValueError: Missing some input keys: {'examples'}

目前，_call方法不处理任何"examples"字段/输入。

generated_cypher = self.cypher_generation_chain.run(
            {"question": question, "schema": self.graph_schema}, callbacks=callbacks
        )

我的建议是重写_call方法并更改以下代码。如果你需要帮助，请告诉我。

generated_cypher = self.cypher_generation_chain.run(
            {"question": question, "schema": self.graph_schema, "examples": inputs["examples"]}, callbacks=callbacks
        )

感谢@RafaelXokito。是的，如果你们能提供补丁，我们将不胜感激。

你好@pierreoberholzer,
这里有一个简单的例子：

class GraphCypherQAChainAux(GraphCypherQAChain):
    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """Generate Cypher statement, use it to look up in the database and answer the question."""
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        callbacks = _run_manager.get_child()
        question = inputs[self.input_key]

        intermediate_steps: List = []

        generated_cypher = self.cypher_generation_chain.run(
            {"question": question, "schema": self.graph_schema, "examples": inputs["examples"]}, callbacks=callbacks
        )

        # Extract Cypher code if it is wrapped in backticks
        generated_cypher = extract_cypher(generated_cypher)

        # Correct Cypher query if enabled
        if self.cypher_query_corrector:
            generated_cypher = self.cypher_query_corrector(generated_cypher)

        _run_manager.on_text("Generated Cypher:", end="\n", verbose=self.verbose)
        _run_manager.on_text(
            generated_cypher, color="green", end="\n", verbose=self.verbose
        )

        intermediate_steps.append({"query": generated_cypher})

        # Retrieve and limit the number of results
        # Generated Cypher can be null if query corrector identifies an invalid schema
        if generated_cypher:
            context = self.graph.query(generated_cypher)[: self.top_k]
        else:
            context = []

        if self.return_direct:
            final_result = context
        else:
            _run_manager.on_text("Full Context:", end="\n", verbose=self.verbose)
            _run_manager.on_text(
                str(context), color="green", end="\n", verbose=self.verbose
            )

            intermediate_steps.append({"context": context})

            result = self.qa_chain(
                {"question": question, "context": context},
                callbacks=callbacks,
            )
            final_result = result[self.qa_chain.output_key]

        chain_result: Dict[str, Any] = {self.output_key: final_result}
        if self.return_intermediate_steps:
            chain_result[INTERMEDIATE_STEPS_KEY] = intermediate_steps

        return chain_result

这里是如何使用它的：

CYPHER_QA_TEMPLATE = """

You're an AI cook formulating Cypher statements to navigate through a recipe database.

Schema: {schema}

Examples: {examples}

Question: {question}

"""

CYPHER_GENERATION_PROMPT = PromptTemplate(
    input_variables=["schema", "examples", "question"],
    template=CYPHER_QA_TEMPLATE)

chain = GraphCypherQAChainAux.from_llm(
    graph=graph, llm=model, verbose=True, validate_cypher=True,
    cypher_prompt=CYPHER_GENERATION_PROMPT, input_key="question")

res = chain.invoke({"examples": examples, "question": question})

请注意，我没有包括模式，因为原始的 _call 方法已经包含了它。
请对此给我你的反馈。

非常感谢@RafaelXokito!您修补的类版本正在工作。
但这是一种相当的解决方法。
包里有任何变化吗？

感谢@pierreoberholzer的反馈！
我正在考虑对cypher_generation_chain中输入的使用方式进行修改。具体来说，我打算动态地连接输入以简化流程：

question = inputs[self.input_key]
args = {
    "question": question,
    "schema": self.graph_schema,
}
args.update(inputs)

intermediate_steps: List = []

generated_cypher = self.cypher_generation_chain.run(
    args, callbacks=callbacks
)

然而，我对这个更改可能对现有方法的用户产生的影响感到担忧。@ccurme,您能就这个提议的修改提供一些见解吗？
谢谢！

你好@pierreoberholzer,
这里有一个简单的例子：

class GraphCypherQAChainAux(GraphCypherQAChain):
    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """Generate Cypher statement, use it to look up in the database and answer the question."""
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        callbacks = _run_manager.get_child()
        question = inputs[self.input_key]

        intermediate_steps: List = []

        generated_cypher = self.cypher_generation_chain.run(
            {"question": question, "schema": self.graph_schema, "examples": inputs["examples"]}, callbacks=callbacks
        )

        # Extract Cypher code if it is wrapped in backticks
        generated_cypher = extract_cypher(generated_cypher)

        # Correct Cypher query if enabled
        if self.cypher_query_corrector:
            generated_cypher = self.cypher_query_corrector(generated_cypher)

        _run_manager.on_text("Generated Cypher:", end="\n", verbose=self.verbose)
        _run_manager.on_text(
            generated_cypher, color="green", end="\n", verbose=self.verbose
        )

        intermediate_steps.append({"query": generated_cypher})

        # Retrieve and limit the number of results
        # Generated Cypher can be null if query corrector identifies an invalid schema
        if generated_cypher:
            context = self.graph.query(generated_cypher)[: self.top_k]
        else:
            context = []

        if self.return_direct:
            final_result = context
        else:
            _run_manager.on_text("Full Context:", end="\n", verbose=self.verbose)
            _run_manager.on_text(
                str(context), color="green", end="\n", verbose=self.verbose
            )

            intermediate_steps.append({"context": context})

            result = self.qa_chain(
                {"question": question, "context": context},
                callbacks=callbacks,
            )
            final_result = result[self.qa_chain.output_key]

        chain_result: Dict[str, Any] = {self.output_key: final_result}
        if self.return_intermediate_steps:
            chain_result[INTERMEDIATE_STEPS_KEY] = intermediate_steps

        return chain_result

这里是如何使用它的：

CYPHER_QA_TEMPLATE = """

You're an AI cook formulating Cypher statements to navigate through a recipe database.

Schema: {schema}

Examples: {examples}

Question: {question}

"""

CYPHER_GENERATION_PROMPT = PromptTemplate(
    input_variables=["schema", "examples", "question"],
    template=CYPHER_QA_TEMPLATE)

chain = GraphCypherQAChainAux.from_llm(
    graph=graph, llm=model, verbose=True, validate_cypher=True,
    cypher_prompt=CYPHER_GENERATION_PROMPT, input_key="question")

res = chain.invoke({"examples": examples, "question": question})

请注意，我没有包括模式，因为原始的 _call 方法已经获取到了它。
请就这个给我你的反馈。
我最终做了一些类似的事情来使其正常工作，以便生成SPARQL。我也在处理TTL文件，所以你还需要重写RdfGraph类。