跳至内容

数据库

数据库工具规范 #

基类: BaseToolSpec, BaseReader

简易数据库工具。

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

参数:

名称 类型 描述 默认值
sql_database Optional[SQLDatabase]

要使用的SQL数据库,包括需要指定的表名。详情请参阅:Ref-Struct-Store

None
engine Optional[Engine]

数据库连接的SQLAlchemy引擎对象。

None
uri Optional[str]

数据库连接的URI。

None
scheme Optional[str]

数据库连接的方案。

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-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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(query: str) -> List[Document]

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

参数:

名称 类型 描述 默认值
query str

一个用于筛选表和行的SQL查询。

required

返回:

类型 描述
List[Document]

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

Source code in llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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端点。

Source code in llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py
106
107
108
109
110
111
112
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(tables: Optional[List[str]] = None) -> str

描述数据库中指定的表。

参数:

名称 类型 描述 默认值
tables List[str]

要检索详细信息的表名列表

None
Source code in llama-index-integrations/tools/llama-index-tools-database/llama_index/tools/database/base.py
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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)