# 从配置文件中获取OpenAI API密钥
openai_api_key = config["OPENAI"]["OPENAI_API_KEY"]
import os

# 将OpenAI API密钥添加到环境变量中
os.environ.update({"OPENAI_API_KEY": openai_api_key})

# 检查配置文件中是否存在"WIKIDATA"部分
# 如果存在，则将"WIKIDATA_USER_AGENT_HEADER"的值赋给wikidata_user_agent_header变量，否则将其赋值为None
wikidata_user_agent_header = (
    None
    if not config.has_section("WIKIDATA")
    else config["WIKIDATA"]["WIKIDATA_USER_AGENT_HEADER"]
)

# 导入os模块
import os

# 设置环境变量LANGCHAIN_HANDLER为"langchain"
os.environ["LANGCHAIN_HANDLER"] = "langchain"

# 设置环境变量LANGCHAIN_SESSION为"default"，确保该会话实际存在
os.environ["LANGCHAIN_SESSION"] = "default"

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


from typing import Optional

import requests


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."

import json
from typing import Any, Dict, List

import requests


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

    # 发送GET请求
    response = requests.get(
        url, headers=headers, params={"query": query, "format": "json"}
    )

    # 检查响应状态码
    if response.status_code != 200:
        return "That query failed. Perhaps you could try a different one?"

    # 获取嵌套值
    results = get_nested_value(response.json(), ["results", "bindings"])

    # 将结果转换为JSON格式并返回
    return json.dumps(results)
以上是一个运行SPARQL查询的函数。该函数使用requests库发送GET请求到指定的URL，并将查询结果转换为JSON格式返回。函数的参数包括查询字符串、URL和用户代理头。如果请求失败，函数会返回一个错误消息。

导入所需的模块和类
import re  # 导入正则表达式模块
from typing import List, Union  # 导入类型提示模块

from langchain.agents import (  # 导入langchain.agents模块中的类
    AgentExecutor,  # 导入AgentExecutor类
    AgentOutputParser,  # 导入AgentOutputParser类
    LLMSingleActionAgent,  # 导入LLMSingleActionAgent类
    Tool,  # 导入Tool类
)
from langchain.chains import LLMChain  # 导入langchain.chains模块中的LLMChain类
from langchain.prompts import StringPromptTemplate  # 导入langchain.prompts模块中的StringPromptTemplate类
from langchain_core.agents import AgentAction, AgentFinish  # 导入langchain_core.agents模块中的AgentAction和AgentFinish类

# 定义代理可以使用的工具来回答用户查询
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="用于从维基库获取结果很有用",
    ),
]

# 设置基本模板
template = """
通过对维基库进行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"\nObservation: {observation}\nThought: "
        # 将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"],  # 指定输入变量
)

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
        )

# 创建一个名为CustomOutputParser的自定义输出解析器对象
output_parser = CustomOutputParser()

# 导入langchain_openai库中的ChatOpenAI类
from langchain_openai import ChatOpenAI

# 创建一个ChatOpenAI对象，使用"gpt-4"模型，设置温度为0
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 创建一个LLM链，包含LLM和一个提示
llm_chain = LLMChain(llm=llm, prompt=prompt)

# 创建一个包含工具名称的列表
tool_names = [tool.name for tool in tools]

# 创建一个LLMSingleActionAgent对象，传入llm_chain、output_parser、stop、allowed_tools参数
agent = LLMSingleActionAgent(
    llm_chain=llm_chain,  # llm_chain参数
    output_parser=output_parser,  # output_parser参数
    stop=["\nObservation:"],  # stop参数，当输出中包含"\nObservation:"时停止
    allowed_tools=tool_names,  # allowed_tools参数，允许使用的工具名称列表
)

# 导入AgentExecutor类
from rasa.core.agent import AgentExecutor

# 创建AgentExecutor对象，并传入agent和tools参数，verbose设置为True
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent, tools=tools, verbose=True
)

# 如果您喜欢使用内联追踪，请取消注释以下行
# agent_executor.agent.llm_chain.verbose = True