Playgrounds

PlaygroundsSubgraphConnectorToolSpec

Bases: GraphQLToolSpec

连接到The Graph分布式网络上的子图，通过Playgrounds API。

属性： spec_functions（列表）：指定工具功能的函数列表。 url（str）：GraphQL请求的端点URL。 headers（字典）：用于GraphQL请求的标头。

Source code in llama_index/tools/playgrounds/subgraph_connector/base.py

class PlaygroundsSubgraphConnectorToolSpec(GraphQLToolSpec):
    """连接到The Graph分布式网络上的子图，通过Playgrounds API。

属性：
    spec_functions（列表）：指定工具功能的函数列表。
    url（str）：GraphQL请求的端点URL。
    headers（字典）：用于GraphQL请求的标头。"""

    spec_functions = ["graphql_request"]

    def __init__(self, identifier: str, api_key: str, use_deployment_id: bool = False):
        """初始化连接器。

Args:
    identifier（str）：子图标识符或部署ID。
    api_key（str）：Playgrounds API的API密钥。
    use_deployment_id（bool）：指示标识符是否为部署ID的标志。默认值为False。
"""
        endpoint = "deployments" if use_deployment_id else "subgraphs"
        self.url = (
            f"https://api.playgrounds.network/v1/proxy/{endpoint}/id/{identifier}"
        )
        self.headers = {
            "Content-Type": "application/json",
            "Playgrounds-Api-Key": api_key,
        }

    def graphql_request(
        self,
        query: str,
        variables: Optional[dict] = None,
        operation_name: Optional[str] = None,
    ) -> Union[dict, str]:
        """发起一个GraphQL查询。

Args:
    query (str): 要执行的GraphQL查询字符串。
    variables (dict, optional): GraphQL查询的变量。默认为None。
    operation_name (str, optional): 如果查询中存在多个操作，则为操作的名称。默认为None。

Returns:
    dict: 如果成功，来自GraphQL服务器的响应。
    str: 如果请求失败，则为错误消息。
"""
        payload = {"query": query.strip()}

        if variables:
            payload["variables"] = variables

        if operation_name:
            payload["operationName"] = operation_name

        try:
            response = requests.post(self.url, headers=self.headers, json=payload)

            # Check if the request was successful
            response.raise_for_status()

            # Return the JSON response
            return response.json()

        except requests.RequestException as e:
            # Handle request errors
            return str(e)
        except ValueError as e:
            # Handle JSON decoding errors
            return f"Error decoding JSON: {e}"

graphql_request

graphql_request(
    query: str,
    variables: Optional[dict] = None,
    operation_name: Optional[str] = None,
) -> Union[dict, str]

发起一个GraphQL查询。

Parameters:

Name	Type	Description	Default
query	str	要执行的GraphQL查询字符串。	required
variables	dict	GraphQL查询的变量。默认为None。	None
operation_name	str	如果查询中存在多个操作，则为操作的名称。默认为None。	None

Returns:

Name	Type	Description
dict	Union[dict, str]	如果成功，来自GraphQL服务器的响应。
str	Union[dict, str]	如果请求失败，则为错误消息。

Source code in llama_index/tools/playgrounds/subgraph_connector/base.py

    def graphql_request(
        self,
        query: str,
        variables: Optional[dict] = None,
        operation_name: Optional[str] = None,
    ) -> Union[dict, str]:
        """发起一个GraphQL查询。

Args:
    query (str): 要执行的GraphQL查询字符串。
    variables (dict, optional): GraphQL查询的变量。默认为None。
    operation_name (str, optional): 如果查询中存在多个操作，则为操作的名称。默认为None。

Returns:
    dict: 如果成功，来自GraphQL服务器的响应。
    str: 如果请求失败，则为错误消息。
"""
        payload = {"query": query.strip()}

        if variables:
            payload["variables"] = variables

        if operation_name:
            payload["operationName"] = operation_name

        try:
            response = requests.post(self.url, headers=self.headers, json=payload)

            # Check if the request was successful
            response.raise_for_status()

            # Return the JSON response
            return response.json()

        except requests.RequestException as e:
            # Handle request errors
            return str(e)
        except ValueError as e:
            # Handle JSON decoding errors
            return f"Error decoding JSON: {e}"

PlaygroundsSubgraphInspectorToolSpec

Bases: GraphQLToolSpec

连接到The Graph的分布式网络上的子图，通过Playgrounds API进行内省。 提供处理和总结内省模式以便易于理解的功能。

属性： spec_functions（列表）：指定工具功能的函数列表。 url（str）：用于GraphQL请求的端点URL。 headers（字典）：用于GraphQL请求的标头。

Source code in llama_index/tools/playgrounds/subgraph_inspector/base.py

class PlaygroundsSubgraphInspectorToolSpec(GraphQLToolSpec):
    """连接到The Graph的分布式网络上的子图，通过Playgrounds API进行内省。
提供处理和总结内省模式以便易于理解的功能。

属性：
    spec_functions（列表）：指定工具功能的函数列表。
    url（str）：用于GraphQL请求的端点URL。
    headers（字典）：用于GraphQL请求的标头。"""

    spec_functions = ["introspect_and_summarize_subgraph"]

    def __init__(self, identifier: str, api_key: str, use_deployment_id: bool = False):
        """初始化与The Graph网络上指定子图的连接。

Args:
    identifier（str）：子图的标识符或部署ID。
    api_key（str）：Playgrounds API的API密钥。
    use_deployment_id（bool）：如果为True，则将标识符视为部署ID。默认值为False。
"""
        self.url = self._generate_url(identifier, use_deployment_id)
        self.headers = {
            "Content-Type": "application/json",
            "Playgrounds-Api-Key": api_key,
        }

    def _generate_url(self, identifier: str, use_deployment_id: bool) -> str:
        """根据标识符和是否为部署ID，生成相应的URL。

Args:
    identifier (str): 子图的标识符或部署ID。
    use_deployment_id (bool): 如果为True，则使用部署ID构建URL。

Returns:
    str: 构建的URL。
"""
        endpoint = "deployments" if use_deployment_id else "subgraphs"
        return f"https://api.playgrounds.network/v1/proxy/{endpoint}/id/{identifier}"

    def introspect_and_summarize_subgraph(self) -> str:
        """分析子图并将其模式总结为文本类别。

返回：
    str：对内省子图模式的文本摘要。
"""
        introspection_query = """
        query {
            __schema {
                types {
                    kind
                    name
                    description
                    enumValues {
                        name
                    }
                    fields {
                        name
                        args {
                            name
                        }
                        type {
                            kind
                            name
                            ofType {
                                name
                            }
                        }
                    }
                }
            }
        }
        """
        response = self._graphql_request(introspection_query)
        if "data" in response:
            result = response["data"]
            processed_subgraph = self._process_subgraph(result)
            return self.subgraph_to_text(processed_subgraph)
        else:
            return "Error during introspection."

    def _graphql_request(self, query: str) -> dict:
        """执行针对子图端点的GraphQL查询。

Args:
    query (str): GraphQL查询字符串。

Returns:
    dict: 来自GraphQL服务器的响应，可能包含数据或错误。
"""
        payload = {"query": query.strip()}
        try:
            response = requests.post(self.url, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            return {"error": str(e)}

    def _process_subgraph(self, result: dict) -> dict:
        """处理基于命名约定的内省子图模式的过程。

Args:
    result（字典）：来自GraphQL查询的内省模式结果。

Returns:
    字典：内省模式的处理表示，分类为特定实体查询、实体列表查询和其他实体。
"""
        processed_subgraph = {
            "specific_entity_queries": {},
            "list_entity_queries": {},
            "other_entities": {},
        }
        for type_ in result["__schema"]["types"]:
            if type_["name"].startswith("__"):
                continue  # Skip meta entities

            entity_name = type_["name"]
            fields, args_required = self._get_fields(type_)
            if fields:
                # Determine category based on naming convention
                if entity_name.endswith("s") and not args_required:
                    processed_subgraph["list_entity_queries"][entity_name] = fields
                elif not entity_name.endswith("s") and args_required:
                    processed_subgraph["specific_entity_queries"][entity_name] = fields
                else:
                    processed_subgraph["other_entities"][entity_name] = fields

        return processed_subgraph

    def _get_fields(self, type_):
        """从内省模式中提取给定类型的相关字段及其详细信息。

Args:
    type_ (dict): 内省模式中的一个类型。

Returns:
    tuple: 包含相关字段列表和一个布尔值，指示字段是否需要参数的元组。
"""
        fields = []
        args_required = False
        for f in type_.get("fields") or []:
            if f["name"] != "__typename" and not (
                f["name"].endswith("_filter")
                or f["name"].endswith("_orderBy")
                or f["name"].islower()
            ):
                field_info = {"name": f["name"]}

                # Check for enum values
                if "enumValues" in f["type"] and f["type"]["enumValues"]:
                    field_info["enumValues"] = [
                        enum_val["name"] for enum_val in f["type"]["enumValues"]
                    ]

                fields.append(field_info)
                if f.get("args") and len(f["args"]) > 0:
                    args_required = True
                if f.get("type") and f["type"].get("fields"):
                    subfields, sub_args_required = self._get_fields(f["type"])
                    fields.extend(subfields)
                    if sub_args_required:
                        args_required = True
        return fields, args_required

    def format_section(
        self, category: str, description: str, example: str, entities: dict
    ) -> str:
        """将子图内省结果的给定部分格式化为可读的字符串格式。

Args:
    category（str）：实体的类别名称。
    description（str）：解释该类别的描述。
    example（str）：与该类别相关的通用GraphQL查询示例。
    entities（dict）：包含与该类别相关的实体及其字段的字典。

Returns:
    str：所提供部分数据的格式化字符串表示形式。
"""
        section = [
            f"Category: {category}",
            f"Description: {description}",
            "Generic Example:",
            example,
            "\nDetailed Breakdown:",
        ]

        for entity, fields in entities.items():
            section.append(f"  Entity: {entity}")
            for field_info in fields:
                field_str = f"    - {field_info['name']}"
                if "enumValues" in field_info:
                    field_str += (
                        f" (Enum values: {', '.join(field_info['enumValues'])})"
                    )
                section.append(field_str)
            section.append("")  # Add a blank line for separation

        section.append("")  # Add another blank line for separation between sections
        return "\n".join(section)

    def subgraph_to_text(self, subgraph: dict) -> str:
        """将处理过的子图表示转换为基于实体类别的文本摘要。

Args:
    subgraph（字典）：内省模式的处理表示，分类为特定实体查询、列表实体查询和其他实体。

Returns:
    str：处理过的子图模式的文本摘要。
"""
        sections = [
            (
                "Specific Entity Queries (Requires Arguments)",
                "These queries target a singular entity and require specific arguments (like an ID) to fetch data.",
                """
            {
                entityName(id: "specific_id") {
                    fieldName1
                    fieldName2
                    ...
                }
            }
            """,
                subgraph["specific_entity_queries"],
            ),
            (
                "List Entity Queries (Optional Arguments)",
                "These queries fetch a list of entities. They don't strictly require arguments but often accept optional parameters for filtering, sorting, and pagination.",
                """
            {
                entityNames(first: 10, orderBy: "someField", orderDirection: "asc") {
                    fieldName1
                    fieldName2
                    ...
                }
            }
            """,
                subgraph["list_entity_queries"],
            ),
            (
                "Other Entities",
                "These are additional entities that may not fit the conventional singular/plural querying pattern of subgraphs.",
                "",
                subgraph["other_entities"],
            ),
        ]

        result_lines = []
        for category, desc, example, entities in sections:
            result_lines.append(self.format_section(category, desc, example, entities))

        return "\n".join(result_lines)

introspect_and_summarize_subgraph

introspect_and_summarize_subgraph() -> str

分析子图并将其模式总结为文本类别。

返回： str：对内省子图模式的文本摘要。

Source code in llama_index/tools/playgrounds/subgraph_inspector/base.py

    def introspect_and_summarize_subgraph(self) -> str:
        """分析子图并将其模式总结为文本类别。

返回：
    str：对内省子图模式的文本摘要。
"""
        introspection_query = """
        query {
            __schema {
                types {
                    kind
                    name
                    description
                    enumValues {
                        name
                    }
                    fields {
                        name
                        args {
                            name
                        }
                        type {
                            kind
                            name
                            ofType {
                                name
                            }
                        }
                    }
                }
            }
        }
        """
        response = self._graphql_request(introspection_query)
        if "data" in response:
            result = response["data"]
            processed_subgraph = self._process_subgraph(result)
            return self.subgraph_to_text(processed_subgraph)
        else:
            return "Error during introspection."

format_section

format_section(
    category: str,
    description: str,
    example: str,
    entities: dict,
) -> str

将子图内省结果的给定部分格式化为可读的字符串格式。

Returns:

Type	Description
str	str：所提供部分数据的格式化字符串表示形式。

Source code in llama_index/tools/playgrounds/subgraph_inspector/base.py

    def format_section(
        self, category: str, description: str, example: str, entities: dict
    ) -> str:
        """将子图内省结果的给定部分格式化为可读的字符串格式。

Args:
    category（str）：实体的类别名称。
    description（str）：解释该类别的描述。
    example（str）：与该类别相关的通用GraphQL查询示例。
    entities（dict）：包含与该类别相关的实体及其字段的字典。

Returns:
    str：所提供部分数据的格式化字符串表示形式。
"""
        section = [
            f"Category: {category}",
            f"Description: {description}",
            "Generic Example:",
            example,
            "\nDetailed Breakdown:",
        ]

        for entity, fields in entities.items():
            section.append(f"  Entity: {entity}")
            for field_info in fields:
                field_str = f"    - {field_info['name']}"
                if "enumValues" in field_info:
                    field_str += (
                        f" (Enum values: {', '.join(field_info['enumValues'])})"
                    )
                section.append(field_str)
            section.append("")  # Add a blank line for separation

        section.append("")  # Add another blank line for separation between sections
        return "\n".join(section)

subgraph_to_text

subgraph_to_text(subgraph: dict) -> str

将处理过的子图表示转换为基于实体类别的文本摘要。

Returns:

Type	Description
str	str：处理过的子图模式的文本摘要。

Source code in llama_index/tools/playgrounds/subgraph_inspector/base.py

    def subgraph_to_text(self, subgraph: dict) -> str:
        """将处理过的子图表示转换为基于实体类别的文本摘要。

Args:
    subgraph（字典）：内省模式的处理表示，分类为特定实体查询、列表实体查询和其他实体。

Returns:
    str：处理过的子图模式的文本摘要。
"""
        sections = [
            (
                "Specific Entity Queries (Requires Arguments)",
                "These queries target a singular entity and require specific arguments (like an ID) to fetch data.",
                """
            {
                entityName(id: "specific_id") {
                    fieldName1
                    fieldName2
                    ...
                }
            }
            """,
                subgraph["specific_entity_queries"],
            ),
            (
                "List Entity Queries (Optional Arguments)",
                "These queries fetch a list of entities. They don't strictly require arguments but often accept optional parameters for filtering, sorting, and pagination.",
                """
            {
                entityNames(first: 10, orderBy: "someField", orderDirection: "asc") {
                    fieldName1
                    fieldName2
                    ...
                }
            }
            """,
                subgraph["list_entity_queries"],
            ),
            (
                "Other Entities",
                "These are additional entities that may not fit the conventional singular/plural querying pattern of subgraphs.",
                "",
                subgraph["other_entities"],
            ),
        ]

        result_lines = []
        for category, desc, example, entities in sections:
            result_lines.append(self.format_section(category, desc, example, entities))

        return "\n".join(result_lines)