使用插件检索的自定义代理¶

# 导入所需的模块
import re
from typing import Union

# 导入自定义模块
from langchain.agents import (
    AgentExecutor,
    AgentOutputParser,
    LLMSingleActionAgent,
)
from langchain.chains import LLMChain
from langchain.prompts import StringPromptTemplate
from langchain_community.agent_toolkits import NLAToolkit
from langchain_community.tools.plugin import AIPlugin
from langchain_core.agents import AgentAction, AgentFinish
from langchain_openai import OpenAI

# 创建OpenAI对象，并设置温度参数为0
llm = OpenAI(temperature=0)

# 定义一个包含多个URL的列表
urls = [
    "https://datasette.io/.well-known/ai-plugin.json",
    "https://api.speak.com/.well-known/ai-plugin.json",
    "https://www.wolframalpha.com/.well-known/ai-plugin.json",
    "https://www.zapier.com/.well-known/ai-plugin.json",
    "https://www.klarna.com/.well-known/ai-plugin.json",
    "https://www.joinmilo.com/.well-known/ai-plugin.json",
    "https://slack.com/.well-known/ai-plugin.json",
    "https://schooldigger.com/.well-known/ai-plugin.json",
]

# 使用列表推导式从URL列表中创建AIPlugin对象的列表
AI_PLUGINS = [AIPlugin.from_url(url) for url in urls]

从langchain_community.vectorstores导入FAISS模块
从langchain_core.documents导入Document模块
从langchain_openai导入OpenAIEmbeddings模块

# 创建一个 retriever 对象，用于检索向量存储中的数据
retriever = vector_store.as_retriever()

def get_tools(query):
    # 获取包含要使用的插件的文档
    docs = retriever.invoke(query)
    # 获取每个插件对应的工具包
    tool_kits = [toolkits_dict[d.metadata["plugin_name"]] for d in docs]
    # 获取工具：每个端点对应一个单独的 NLAChain
    tools = []
    for tk in tool_kits:
        tools.extend(tk.nla_tools)
    return tools

# 设置基本模板
template = """以海盗的口吻回答以下问题，尽力而为。你可以使用以下工具：

{tools}

使用以下格式：

问题：你必须回答的输入问题
思考：你应该始终考虑该怎么做
行动：要采取的行动，应该是[{tool_names}]之一
行动输入：行动的输入
观察：行动的结果
...（这个思考/行动/行动输入/观察可以重复N次）
思考：我现在知道最终答案了
最终答案：原始输入问题的最终答案

开始！记住在给出最终答案时要以海盗的口吻说话。使用大量的"Arg"

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

from typing import Callable


# 设置一个提示模板
class CustomPromptTemplate(StringPromptTemplate):
    # 要使用的模板
    template: str
    ############## 新增 ######################
    # 可用工具的列表
    tools_getter: Callable

    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 = self.tools_getter(kwargs["input"])
        # 从提供的工具列表创建一个tools变量
        kwargs["tools"] = "\n".join(
            [f"{tool.name}: {tool.description}" for tool in tools]
        )
        # 为提供的工具创建一个工具名称列表
        kwargs["tool_names"] = ", ".join([tool.name for tool in tools])
        return self.template.format(**kwargs)

prompt = CustomPromptTemplate(
    template=template,  # 使用给定的模板创建自定义提示
    tools_getter=get_tools,  # 使用get_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\s*\d*\s*:(.*?)\nAction\s*\d*\s*Input\s*\d*\s*:[\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()  # 创建一个名为output_parser的对象，使用CustomOutputParser类进行初始化

# 创建一个OpenAI对象，设置温度为0
llm = OpenAI(temperature=0)

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

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

# 创建一个LLMSingleActionAgent对象
# 参数说明：
# llm_chain: LLM链
# output_parser: 输出解析器
# stop: 停止标志，当遇到"\nObservation:"时停止
# allowed_tools: 允许使用的工具名称列表
agent = LLMSingleActionAgent(
    llm_chain=llm_chain,
    output_parser=output_parser,
    stop=["\nObservation:"],
    allowed_tools=tool_names,
)

# 导入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
)