ElasticsearchDatabaseChain Class — langchain Architecture

Architecture documentation for the ElasticsearchDatabaseChain class in base.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  3a699a2b_5ca7_babd_1b5e_e76699b2210a["ElasticsearchDatabaseChain"]
  097a4781_5519_0b5d_6244_98c64eadc0d6["Chain"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|extends| 097a4781_5519_0b5d_6244_98c64eadc0d6
  4c0291a0_dacd_3a35_a892_6bde5e7778eb["base.py"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|defined in| 4c0291a0_dacd_3a35_a892_6bde5e7778eb
  8d6d0452_ceec_15e4_16e3_8f4b10f6ef63["_validate_indices()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| 8d6d0452_ceec_15e4_16e3_8f4b10f6ef63
  dbe3d9b7_631d_a2ff_04de_47849b37a8b3["input_keys()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| dbe3d9b7_631d_a2ff_04de_47849b37a8b3
  acd15cbe_26bb_f792_cb60_0cac36466bbb["output_keys()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| acd15cbe_26bb_f792_cb60_0cac36466bbb
  e9c35e4f_3e75_dbcd_8b6a_7b6eb9a9c394["_list_indices()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| e9c35e4f_3e75_dbcd_8b6a_7b6eb9a9c394
  a666aaf2_7c4a_7350_db34_3270f49691ef["_get_indices_infos()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| a666aaf2_7c4a_7350_db34_3270f49691ef
  de3258d6_d683_3e31_b6f8_ba6213417a67["_search()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| de3258d6_d683_3e31_b6f8_ba6213417a67
  d38da816_4fa9_fe6e_5ab1_e931d1edbc93["_call()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| d38da816_4fa9_fe6e_5ab1_e931d1edbc93
  9fd3a185_26fd_681e_e5da_3bc33f4a7a9d["_chain_type()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| 9fd3a185_26fd_681e_e5da_3bc33f4a7a9d
  57aa9204_a4f8_7532_b024_d017241ba41e["from_llm()"]
  3a699a2b_5ca7_babd_1b5e_e76699b2210a -->|method| 57aa9204_a4f8_7532_b024_d017241ba41e

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/elasticsearch_database/base.py lines 28–208

class ElasticsearchDatabaseChain(Chain):
    """Chain for interacting with Elasticsearch Database.

    Example:
        ```python
        from langchain_classic.chains import ElasticsearchDatabaseChain
        from langchain_openai import OpenAI
        from elasticsearch import Elasticsearch

        database = Elasticsearch("http://localhost:9200")
        db_chain = ElasticsearchDatabaseChain.from_llm(OpenAI(), database)
        ```
    """

    query_chain: Runnable
    """Chain for creating the ES query."""
    answer_chain: Runnable
    """Chain for answering the user question."""
    database: Any = None
    """Elasticsearch database to connect to of type elasticsearch.Elasticsearch."""
    top_k: int = 10
    """Number of results to return from the query"""
    ignore_indices: list[str] | None = None
    include_indices: list[str] | None = None
    input_key: str = "question"
    output_key: str = "result"
    sample_documents_in_index_info: int = 3
    return_intermediate_steps: bool = False
    """Whether or not to return the intermediate steps along with the final answer."""

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        extra="forbid",
    )

    @model_validator(mode="after")
    def _validate_indices(self) -> Self:
        if self.include_indices and self.ignore_indices:
            msg = "Cannot specify both 'include_indices' and 'ignore_indices'."
            raise ValueError(msg)
        return self

    @property
    def input_keys(self) -> list[str]:
        """Return the singular input key."""
        return [self.input_key]

    @property
    def output_keys(self) -> list[str]:
        """Return the singular output key."""
        if not self.return_intermediate_steps:
            return [self.output_key]
        return [self.output_key, INTERMEDIATE_STEPS_KEY]

    def _list_indices(self) -> list[str]:
        all_indices = [
            index["index"] for index in self.database.cat.indices(format="json")
        ]

        if self.include_indices:
            all_indices = [i for i in all_indices if i in self.include_indices]
        if self.ignore_indices:
            all_indices = [i for i in all_indices if i not in self.ignore_indices]

        return all_indices

    def _get_indices_infos(self, indices: list[str]) -> str:
        mappings = self.database.indices.get_mapping(index=",".join(indices))
        if self.sample_documents_in_index_info > 0:
            for k, v in mappings.items():
                hits = self.database.search(
                    index=k,
                    query={"match_all": {}},
                    size=self.sample_documents_in_index_info,
                )["hits"]["hits"]
                hits = [str(hit["_source"]) for hit in hits]
                mappings[k]["mappings"] = str(v) + "\n\n/*\n" + "\n".join(hits) + "\n*/"
        return "\n\n".join(
            [
                "Mapping for index {}:\n{}".format(index, mappings[index]["mappings"])
                for index in mappings