Source code for langchain.chains.graph_qa.cypher
"""Question answering over a graph."""
from __future__ import annotations
import re
from typing import Any, Dict, List, Optional
from langchain.callbacks.manager import CallbackManagerForChainRun
from langchain.chains.base import Chain
from langchain.chains.graph_qa.prompts import CYPHER_GENERATION_PROMPT, CYPHER_QA_PROMPT
from langchain.chains.llm import LLMChain
from langchain.graphs.neo4j_graph import Neo4jGraph
from langchain.pydantic_v1 import Field
from langchain.schema import BasePromptTemplate
from langchain.schema.language_model import BaseLanguageModel
INTERMEDIATE_STEPS_KEY = "intermediate_steps"
[docs]def extract_cypher(text: str) -> str:
    """Extract Cypher code from a text.
    Args:
        text: Text to extract Cypher code from.
    Returns:
        Cypher code extracted from the text.
    """
    # The pattern to find Cypher code enclosed in triple backticks
    pattern = r"```(.*?)```"
    # Find all matches in the input text
    matches = re.findall(pattern, text, re.DOTALL)
    return matches[0] if matches else text
[docs]def construct_schema(
    structured_schema: Dict[str, Any],
    include_types: List[str],
    exclude_types: List[str],
) -> str:
    """Filter the schema based on included or excluded types"""
    def filter_func(x: str) -> bool:
        return x in include_types if include_types else x not in exclude_types
    filtered_schema = {
        "node_props": {
            k: v
            for k, v in structured_schema.get("node_props", {}).items()