Source code for langchain_community.chains.graph_qa.base

"""在图上的问答。"""
from __future__ import annotations

from typing import Any, Dict, List, Optional

from langchain.chains.base import Chain
from langchain.chains.llm import LLMChain
from langchain_core.callbacks.manager import CallbackManagerForChainRun
from langchain_core.language_models import BaseLanguageModel
from langchain_core.prompts import BasePromptTemplate
from langchain_core.pydantic_v1 import Field

from langchain_community.chains.graph_qa.prompts import (
    ENTITY_EXTRACTION_PROMPT,
    GRAPH_QA_PROMPT,
)
from langchain_community.graphs.networkx_graph import NetworkxEntityGraph, get_entities


class GraphQAChain(Chain):
    """用于针对图形进行问答的链。

    *安全提示*：确保数据库连接使用的凭据范围狭窄，仅包括必要的权限。
        如果未这样做，可能会导致数据损坏或丢失，因为调用代码可能尝试执行会导致删除、变异数据或读取敏感数据（如果数据库中存在此类数据）的命令。
        防范这种负面结果的最佳方法是（视情况）限制授予此工具使用的凭据的权限。

        有关更多信息，请参见 https://python.langchain.com/docs/security。"""

    graph: NetworkxEntityGraph = Field(exclude=True)
    entity_extraction_chain: LLMChain
    qa_chain: LLMChain
    input_key: str = "query"  #: :meta private:
    output_key: str = "result"  #: :meta private:

    @property
    def input_keys(self) -> List[str]:
        """输入键。

:元数据 私有:
"""
        return [self.input_key]

    @property
    def output_keys(self) -> List[str]:
        """输出键。

:元数据 私有:
"""
        _output_keys = [self.output_key]
        return _output_keys

    @classmethod
    def from_llm(
        cls,
        llm: BaseLanguageModel,
        qa_prompt: BasePromptTemplate = GRAPH_QA_PROMPT,
        entity_prompt: BasePromptTemplate = ENTITY_EXTRACTION_PROMPT,
        **kwargs: Any,
    ) -> GraphQAChain:
        """从LLM初始化。"""
        qa_chain = LLMChain(llm=llm, prompt=qa_prompt)
        entity_chain = LLMChain(llm=llm, prompt=entity_prompt)

        return cls(
            qa_chain=qa_chain,
            entity_extraction_chain=entity_chain,
            **kwargs,
        )

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, str]:
        """提取实体，查找信息并回答问题。"""
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        question = inputs[self.input_key]

        entity_string = self.entity_extraction_chain.run(question)

        _run_manager.on_text("Entities Extracted:", end="\n", verbose=self.verbose)
        _run_manager.on_text(
            entity_string, color="green", end="\n", verbose=self.verbose
        )
        entities = get_entities(entity_string)
        context = ""
        all_triplets = []
        for entity in entities:
            all_triplets.extend(self.graph.get_entity_knowledge(entity))
        context = "\n".join(all_triplets)
        _run_manager.on_text("Full Context:", end="\n", verbose=self.verbose)
        _run_manager.on_text(context, color="green", end="\n", verbose=self.verbose)
        result = self.qa_chain(
            {"question": question, "context": context},
            callbacks=_run_manager.get_child(),
        )
        return {self.output_key: result[self.qa_chain.output_key]}