数据库

DatabaseToolSpec

基类: BaseToolSpec, BaseReader

简单的数据库工具。

将每一行连接成 LlamaIndex 使用的 Document。

参数

名称	类型	描述	默认值
sql_database	Optional[SQLDatabase]	要使用的 SQL 数据库，包括要指定的表名。详情请参阅 :ref:Ref-Struct-Store。	无
engine	Optional[Engine]	数据库连接的 SQLAlchemy Engine 对象。	无
uri	Optional[str]	数据库连接的 uri。	无
scheme	Optional[str]	数据库连接的 scheme。	无
host	Optional[str]	数据库连接的 host。	无
port	Optional[int]	数据库连接的 port。	无
user	Optional[str]	数据库连接的 user。	无
password	Optional[str]	数据库连接的 password。	无
dbname	Optional[str]	数据库连接的 dbname。	无

源代码位于 llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py

class DatabaseToolSpec(BaseToolSpec, BaseReader):
    """
    Simple Database tool.

    Concatenates each row into Document used by LlamaIndex.

    Args:
        sql_database (Optional[SQLDatabase]): SQL database to use,
            including table names to specify.
            See :ref:`Ref-Struct-Store` for more details.

        OR

        engine (Optional[Engine]): SQLAlchemy Engine object of the database connection.

        OR

        uri (Optional[str]): uri of the database connection.

        OR

        scheme (Optional[str]): scheme of the database connection.
        host (Optional[str]): host of the database connection.
        port (Optional[int]): port of the database connection.
        user (Optional[str]): user of the database connection.
        password (Optional[str]): password of the database connection.
        dbname (Optional[str]): dbname of the database connection.

    """

    spec_functions = ["load_data", "describe_tables", "list_tables"]

    def __init__(
        self,
        sql_database: Optional[SQLDatabase] = None,
        engine: Optional[Engine] = None,
        uri: Optional[str] = None,
        scheme: Optional[str] = None,
        host: Optional[str] = None,
        port: Optional[str] = None,
        user: Optional[str] = None,
        password: Optional[str] = None,
        dbname: Optional[str] = None,
        *args: Optional[Any],
        **kwargs: Optional[Any],
    ) -> None:
        """Initialize with parameters."""
        if sql_database:
            self.sql_database = sql_database
        elif engine:
            self.sql_database = SQLDatabase(engine, *args, **kwargs)
        elif uri:
            self.uri = uri
            self.sql_database = SQLDatabase.from_uri(uri, *args, **kwargs)
        elif scheme and host and port and user and password and dbname:
            uri = f"{scheme}://{user}:{password}@{host}:{port}/{dbname}"
            self.uri = uri
            self.sql_database = SQLDatabase.from_uri(uri, *args, **kwargs)
        else:
            raise ValueError(
                "You must provide either a SQLDatabase, "
                "a SQL Alchemy Engine, a valid connection URI, or a valid "
                "set of credentials."
            )
        self._metadata = MetaData()
        self._metadata.reflect(bind=self.sql_database.engine)

    def load_data(self, query: str) -> List[Document]:
        """
        Query and load data from the Database, returning a list of Documents.

        Args:
            query (str): an SQL query to filter tables and rows.

        Returns:
            List[Document]: A list of Document objects.

        """
        documents = []
        with self.sql_database.engine.connect() as connection:
            if query is None:
                raise ValueError("A query parameter is necessary to filter the data")
            else:
                result = connection.execute(text(query))

            for item in result.fetchall():
                # fetch each item
                doc_str = ", ".join([str(entry) for entry in item])
                documents.append(Document(text=doc_str))
        return documents

    def list_tables(self) -> List[str]:
        """
        Returns a list of available tables in the database.
        To retrieve details about the columns of specific tables, use
        the describe_tables endpoint.
        """
        return [x.name for x in self._metadata.sorted_tables]

    def describe_tables(self, tables: Optional[List[str]] = None) -> str:
        """
        Describes the specified tables in the database.

        Args:
            tables (List[str]): A list of table names to retrieve details about

        """
        table_names = tables or [table.name for table in self._metadata.sorted_tables]
        table_schemas = []

        for table_name in table_names:
            table = next(
                (
                    table
                    for table in self._metadata.sorted_tables
                    if table.name == table_name
                ),
                None,
            )
            if table is None:
                raise NoSuchTableError(f"Table '{table_name}' does not exist.")
            schema = str(CreateTable(table).compile(self.sql_database._engine))
            table_schemas.append(f"{schema}\n")

        return "\n".join(table_schemas)

load_data

load_data(query: str) -> List[Document]

从数据库查询并加载数据，返回 Document 列表。

参数

名称	类型	描述	默认值
query	str	用于过滤表和行的 SQL 查询。	必需

返回

类型	描述
List[Document]	List[Document]：Document 对象列表。

源代码位于 llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py

def load_data(self, query: str) -> List[Document]:
    """
    Query and load data from the Database, returning a list of Documents.

    Args:
        query (str): an SQL query to filter tables and rows.

    Returns:
        List[Document]: A list of Document objects.

    """
    documents = []
    with self.sql_database.engine.connect() as connection:
        if query is None:
            raise ValueError("A query parameter is necessary to filter the data")
        else:
            result = connection.execute(text(query))

        for item in result.fetchall():
            # fetch each item
            doc_str = ", ".join([str(entry) for entry in item])
            documents.append(Document(text=doc_str))
    return documents

list_tables

list_tables() -> List[str]

返回数据库中可用表的列表。要检索特定表的列详细信息，请使用 describe_tables 端点。

源代码位于 llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py

def list_tables(self) -> List[str]:
    """
    Returns a list of available tables in the database.
    To retrieve details about the columns of specific tables, use
    the describe_tables endpoint.
    """
    return [x.name for x in self._metadata.sorted_tables]

describe_tables

describe_tables(tables: Optional[List[str]] = None) -> str

描述数据库中指定的表。

参数

名称	类型	描述	默认值
tables	List[str]	要检索详细信息的表名列表	无

源代码位于 llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py

def describe_tables(self, tables: Optional[List[str]] = None) -> str:
    """
    Describes the specified tables in the database.

    Args:
        tables (List[str]): A list of table names to retrieve details about

    """
    table_names = tables or [table.name for table in self._metadata.sorted_tables]
    table_schemas = []

    for table_name in table_names:
        table = next(
            (
                table
                for table in self._metadata.sorted_tables
                if table.name == table_name
            ),
            None,
        )
        if table is None:
            raise NoSuchTableError(f"Table '{table_name}' does not exist.")
        schema = str(CreateTable(table).compile(self.sql_database._engine))
        table_schemas.append(f"{schema}\n")

    return "\n".join(table_schemas)