Database

DatabaseToolSpec

Bases: BaseToolSpec, BaseReader

简单的数据库工具。

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

Parameters:

Name	Type	Description	Default
sql_database	Optional[SQLDatabase]	要使用的SQL数据库，包括要指定的表名。 有关更多详细信息，请参见:ref:Ref-Struct-Store。	None
engine	Optional[Engine]	数据库连接的SQLAlchemy Engine对象。	None
uri	Optional[str]	数据库连接的uri。	None
scheme	Optional[str]	数据库连接的scheme。	None
host	Optional[str]	数据库连接的主机。	None
port	Optional[int]	数据库连接的端口。	None
user	Optional[str]	数据库连接的用户。	None
password	Optional[str]	数据库连接的密码。	None
dbname	Optional[str]	数据库连接的dbname。	None

Source code in llama_index/tools/database/base.py

class DatabaseToolSpec(BaseToolSpec, BaseReader):
    """简单的数据库工具。

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

Args:
    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]): 数据库连接的主机。
    port (Optional[int]): 数据库连接的端口。
    user (Optional[str]): 数据库连接的用户。
    password (Optional[str]): 数据库连接的密码。
    dbname (Optional[str]): 数据库连接的dbname。"""

    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:
        """使用参数进行初始化。"""
        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]:
        """从数据库中查询并加载数据，返回一个文档列表。

Args:
    query (str): 用于过滤表和行的SQL查询。

Returns:
    List[Document]: 一个Document对象的列表。
"""
        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]:
        """返回数据库中可用表的列表。
要检索特定表的列的详细信息，请使用describe_tables端点。
"""
        return [x.name for x in self._metadata.sorted_tables]

    def describe_tables(self, tables: Optional[List[str]] = None) -> str:
        """描述数据库中指定的表格。

Args:
    tables（List[str]）：要获取详细信息的表格名称列表
"""
        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]

从数据库中查询并加载数据，返回一个文档列表。

Parameters:

Name	Type	Description	Default
query	str	用于过滤表和行的SQL查询。	required

Returns:

Type	Description
List[Document]	List[Document]: 一个Document对象的列表。

Source code in llama_index/tools/database/base.py

    def load_data(self, query: str) -> List[Document]:
        """从数据库中查询并加载数据，返回一个文档列表。

Args:
    query (str): 用于过滤表和行的SQL查询。

Returns:
    List[Document]: 一个Document对象的列表。
"""
        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端点。

Source code in llama_index/tools/database/base.py

    def list_tables(self) -> List[str]:
        """返回数据库中可用表的列表。
要检索特定表的列的详细信息，请使用describe_tables端点。
"""
        return [x.name for x in self._metadata.sorted_tables]

describe_tables

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

描述数据库中指定的表格。

Source code in llama_index/tools/database/base.py

    def describe_tables(self, tables: Optional[List[str]] = None) -> str:
        """描述数据库中指定的表格。

Args:
    tables（List[str]）：要获取详细信息的表格名称列表
"""
        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)