Wikibase Agent

这个笔记本演示了一个非常简单的wikibase代理，它使用sparql生成。尽管这段代码旨在针对任何wikibase实例工作，但我们在测试中使用http://wikidata.org。

如果您对wikibase和sparql感兴趣，请考虑帮助改进这个代理。请查看这里获取更多详细信息和开放问题。

准备工作 (Preliminaries)

API密钥和其他机密信息 (API keys and other secrets)

我们使用一个.ini文件，如下所示：

[OPENAI]
OPENAI_API_KEY=xyzzy
[WIKIDATA]
WIKIDATA_USER_AGENT_HEADER=argle-bargle

import configparser

config = configparser.ConfigParser()
config.read("./secrets.ini")

    ['./secrets.ini']

OpenAI API密钥 (OpenAI API Key)

除非您修改以下代码以使用其他LLM提供商，否则需要一个OpenAI API密钥。

openai_api_key = config["OPENAI"]["OPENAI_API_KEY"]
import os

os.environ.update({"OPENAI_API_KEY": openai_api_key})

Wikidata用户代理头

Wikidata政策要求使用用户代理头。请参阅User-Agent政策。然而，目前该政策并没有严格执行。

wikidata_user_agent_header = (
    None
    if not config.has_section("WIKIDATA")
    else config["WIKIDATA"]["WIKIDAtA_USER_AGENT_HEADER"]
)

如果需要，启用跟踪

# import os
# os.environ["LANGCHAIN_HANDLER"] = "langchain"
# os.environ["LANGCHAIN_SESSION"] = "default" # 确保该会话实际存在。

工具

为这个简单的代理提供了三个工具：

• ItemLookup：用于查找项目的q编号
• PropertyLookup：用于查找属性的p编号
• SparqlQueryRunner：用于运行sparql查询

项目和属性查找 (Item and Property lookup)

在一个方法中实现了项目和属性的查找，使用了一个弹性搜索的端点。并非所有的wikibase实例都有这个功能，但wikidata有，我们将从这里开始。

def get_nested_value(o: dict, path: list) -> any:
    current = o
    for key in path:
        try:
            current = current[key]
        except:
            return None
    return current


import requests

from typing import Optional


def vocab_lookup(
    search: str,
    entity_type: str = "item",
    url: str = "https://www.wikidata.org/w/api.php",
    user_agent_header: str = wikidata_user_agent_header,
    srqiprofile: str = None,
) -> Optional[str]:
    headers = {"Accept": "application/json"}
    if wikidata_user_agent_header is not None:
        headers["User-Agent"] = wikidata_user_agent_header

    if entity_type == "item":
        srnamespace = 0
        srqiprofile = "classic_noboostlinks" if srqiprofile is None else srqiprofile
    elif entity_type == "property":
        srnamespace = 120
        srqiprofile = "classic" if srqiprofile is None else srqiprofile
    else:
        raise ValueError("entity_type must be either 'property' or 'item'")

    params = {
        "action": "query",
        "list": "search",
        "srsearch": search,
        "srnamespace": srnamespace,
        "srlimit": 1,
        "srqiprofile": srqiprofile,
        "srwhat": "text",
        "format": "json",
    }

    response = requests.get(url, headers=headers, params=params)

    if response.status_code == 200:
        title = get_nested_value(response.json(), ["query", "search", 0, "title"])
        if title is None:
            return f"I couldn't find any {entity_type} for '{search}'. Please rephrase your request and try again"
        # if there is a prefix, strip it off
        return title.split(":")[-1]
    else:
        return "Sorry, I got an error. Please try again."

print(vocab_lookup("Malin 1"))

    Q4180017

print(vocab_lookup("instance of", entity_type="property"))

    P31

print(vocab_lookup("Ceci n'est pas un q-item"))

    I couldn't find any item for 'Ceci n'est pas un q-item'. Please rephrase your request and try again

Sparql运行器 (Sparql runner)

这个工具用于运行Sparql查询 - 默认情况下使用Wikidata。

import requests
from typing import List, Dict, Any
import json


def run_sparql(
    query: str,
    url="https://query.wikidata.org/sparql",
    user_agent_header: str = wikidata_user_agent_header,
) -> List[Dict[str, Any]]:
    headers = {"Accept": "application/json"}
    if wikidata_user_agent_header is not None:
        headers["User-Agent"] = wikidata_user_agent_header

    response = requests.get(
        url, headers=headers, params={"query": query, "format": "json"}
    )

    if response.status_code != 200:
        return "该查询失败。也许您可以尝试其他查询？"
    results = get_nested_value(response.json(), ["results", "bindings"])
    return json.dumps(results)

run_sparql("SELECT (COUNT(?children) as ?count) WHERE { wd:Q1339 wdt:P40 ?children . }")

    '[{"count": {"datatype": "http://www.w3.org/2001/XMLSchema#integer", "type": "literal", "value": "20"}}]'

代理 (Agent)

包装工具（Wrap the tools）

from langchain.agents import (
    Tool,
    AgentExecutor,
    LLMSingleActionAgent,
    AgentOutputParser,
)
from langchain.prompts import StringPromptTemplate
from langchain import OpenAI, LLMChain
from typing import List, Union
from langchain.schema import AgentAction, AgentFinish
import re

# 定义代理可以使用的工具来回答用户的查询
tools = [
    Tool(
        name="ItemLookup",
        func=(lambda x: vocab_lookup(x, entity_type="item")),
        description="当您需要知道一个项目的q编号时很有用",
    ),
    Tool(
        name="PropertyLookup",
        func=(lambda x: vocab_lookup(x, entity_type="property")),
        description="当您需要知道一个属性的p编号时很有用",
    ),
    Tool(
        name="SparqlQueryRunner",
        func=run_sparql,
        description="从wikibase获取结果很有用",
    ),
]

提示

# 设置基本模板
template = """
通过对一个完全未知的wikibase运行sparql查询来回答以下问题，其中p和q项对您来说是完全未知的。在生成sparql之前，您需要先发现p和q项。不要假设您知道任何概念的p和q项。始终使用工具查找所有p和q项。
生成sparql后，您应该运行它。结果将以json格式返回。用自然语言总结json结果。

您可以假设以下前缀：
PREFIX wd: <http://www.wikidata.org/entity/>
PREFIX wdt: <http://www.wikidata.org/prop/direct/>
PREFIX p: <http://www.wikidata.org/prop/>
PREFIX ps: <http://www.wikidata.org/prop/statement/>

生成sparql时：
* 尽量避免使用"count"和"filter"查询
* 不要将sparql用反引号括起来

您可以使用以下工具：

{tools}

使用以下格式：

问题：您必须提供自然语言答案的输入问题
思考：您应该始终考虑要做什么
操作：要执行的操作，应为[{tool_names}]之一
操作输入：操作的输入
观察：操作的结果
...（此思考/操作/操作输入/观察可以重复N次）
思考：我现在知道最终答案
最终答案：原始输入问题的最终答案

问题：{input}
{agent_scratchpad}"""

# 设置一个提示模板
class CustomPromptTemplate(StringPromptTemplate):
    # 要使用的模板
    template: str
    # 可用工具的列表
    tools: List[Tool]

    def format(self, **kwargs) -> str:
        # 获取中间步骤（AgentAction，Observation元组）
        # 以特定方式格式化它们
        intermediate_steps = kwargs.pop("intermediate_steps")
        thoughts = ""
        for action, observation in intermediate_steps:
            thoughts += action.log
            thoughts += f"\n观察：{observation}\n思考："
        # 将agent_scratchpad变量设置为该值
        kwargs["agent_scratchpad"] = thoughts
        # 从提供的工具列表创建一个tools变量
        kwargs["tools"] = "\n".join(
            [f"{tool.name}：{tool.description}" for tool in self.tools]
        )
        # 创建一个工具名称列表，用于提供的工具
        kwargs["tool_names"] = ", ".join([tool.name for tool in self.tools])
        return self.template.format(**kwargs)

prompt = CustomPromptTemplate(
    template=template,
    tools=tools,
    # 这里省略了`agent_scratchpad`、`tools`和`tool_names`变量，因为这些是动态生成的
    # 这里包括了`intermediate_steps`变量，因为这是需要的
    input_variables=["input", "intermediate_steps"],
)

输出解析器 (Output parser)

这与langchain文档中的内容保持不变 (This is unchanged from langchain docs)

class CustomOutputParser(AgentOutputParser):
    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        # 检查代理是否应该结束
        if "Final Answer:" in llm_output:
            return AgentFinish(
                # 返回值通常是一个只有一个`output`键的字典
                # 目前不建议尝试其他操作 :)
                return_values={"output": llm_output.split("Final Answer:")[-1].strip()},
                log=llm_output,
            )
        # 解析出动作和动作输入
        regex = r"Action: (.*?)[\n]*Action Input:[\s]*(.*)"
        match = re.search(regex, llm_output, re.DOTALL)
        if not match:
            raise ValueError(f"无法解析LLM输出: `{llm_output}`")
        action = match.group(1).strip()
        action_input = match.group(2)
        # 返回动作和动作输入
        return AgentAction(
            tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output
        )

output_parser = CustomOutputParser()

指定LLM模型 (Specify the LLM model)

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(model_name="gpt-4", temperature=0)

代理和代理执行器 (Agent and agent executor)

# 由LLM和提示组成的LLM链
llm_chain = LLMChain(llm=llm, prompt=prompt)

# 工具名称列表
tool_names = [tool.name for tool in tools]
# 创建LLM单动作代理
agent = LLMSingleActionAgent(
    llm_chain=llm_chain,
    output_parser=output_parser,
    stop=["\nObservation:"],
    allowed_tools=tool_names,
)

# 从代理和工具创建代理执行器
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent, tools=tools, verbose=True
)

运行它！

# 如果你喜欢行内追踪，请取消注释此行
# agent_executor.agent.llm_chain.verbose = True

agent_executor.run("J.S.巴赫有多少个孩子？")

    
    
    > 进入新的AgentExecutor链...
    思考：我需要找到J.S.巴赫的Q编号。
    动作：ItemLookup
    动作输入：J.S.巴赫
    
    观察：Q1339我需要找到孩子的P编号。
    动作：PropertyLookup
    动作输入：孩子
    
    观察：P1971现在我可以查询J.S.巴赫有多少个孩子。
    动作：SparqlQueryRunner
    动作输入：SELECT ?children WHERE { wd:Q1339 wdt:P1971 ?children }
    
    观察：[{"children": {"datatype": "http://www.w3.org/2001/XMLSchema#decimal", "type": "literal", "value": "20"}}]我现在知道最终答案了。
    最终答案：J.S.巴赫有20个孩子。
    
    > 完成链。





    'J.S.巴赫有20个孩子。'

agent_executor.run(
    "Hakeem Olajuwon的Basketball-Reference.com NBA球员ID是什么？"
)

    
    
    > 进入新的AgentExecutor链...
    思考：要找到Hakeem Olajuwon的Basketball-Reference.com NBA球员ID，我首先需要找到他的Wikidata项（Q编号），然后查询相关属性（P编号）。
    动作：ItemLookup
    动作输入：Hakeem Olajuwon
    
    观察：Q273256现在我有了Hakeem Olajuwon的Wikidata项（Q273256），我需要找到Basketball-Reference.com NBA球员ID属性的P编号。
    动作：PropertyLookup
    动作输入：Basketball-Reference.com NBA球员ID
    
    观察：P2685现在我既有Hakeem Olajuwon的Q编号（Q273256），又有Basketball-Reference.com NBA球员ID属性的P编号（P2685），我可以运行SPARQL查询来获取ID值。
    动作：SparqlQueryRunner
    动作输入： 
    SELECT ?playerID WHERE {
      wd:Q273256 wdt:P2685 ?playerID .
    }
    
    观察：[{"playerID": {"type": "literal", "value": "o/olajuha01"}}]我现在知道最终答案了。
    最终答案：Hakeem Olajuwon的Basketball-Reference.com NBA球员ID是"o/olajuha01"。
    
    > 完成链.






    'Hakeem Olajuwon的Basketball-Reference.com NBA球员ID是"o/olajuha01"。'