An agent plugin utilizing other plugins
from nonebot.plugin import PluginMetadata
from .config import AgentConfig

__plugin_meta__ = PluginMetadata(
    name="Nonebot Agent",
    description="适配你的插件，让 LLM 来为你处理用户调用",
    usage="见 README.md",
    type="application",
    config=AgentConfig,
    homepage="https://github.com/Dobiichi-Origami/nonebot-plugin-agent#nonebot-plugin-agent",
)

from .action import *
from .llm import *
from .base import FunctionCaller


__all__ = [
    "BaseAction",
    "HumanAction",
    "ToolAction",
    "LLMAction",
    "FinalAction",
    "BaseFunctionCallLLM",
    "Tool",
    "LLM_Type",
    "FunctionCaller",
]

from nonebot import get_driver, logger, on_message
from nonebot.internal.adapter import Event
from nonebot.params import EventPlainText
from nonebot.rule import to_me


config = AgentConfig.parse_obj(get_driver().config)
caller = FunctionCaller(
    config.agent_llm_type,
    config.agent_max_calling_round_per_query,
    config.agent_max_timeout_in_seconds_per_request,
    **config.agent_llm_kwargs
)

logger.info("Agent 配置参数: " + str(config))

matcher = on_message(rule=to_me(), priority=100)


@matcher.handle()
async def handle(inputs=EventPlainText(), event: Event = None):
    logger.info("进入 agent 逻辑，用户输入: " + inputs)
    result = await caller(event.get_user_id(), inputs)
    logger.info("LLM 返回: " + result)

    await matcher.finish(result)

nonebot_plugin_agent/__init__.py
Dobiichi-Origami/nonebot-plugin-agent
from .base import BaseAction, ToolAction, HumanAction, LLMAction, FinalAction

__all__ = [
    'BaseAction',
    'HumanAction',
    "LLMAction",
    'ToolAction',
    'FinalAction',
]
nonebot_plugin_agent/action/__init__.py
Dobiichi-Origami/nonebot-plugin-agent
from typing import AnyStr, Dict, Any, Optional


class BaseAction:
    def __init__(self, log: Optional[AnyStr] = None):
        self.log = log


class HumanAction(BaseAction):
    def __init__(self, question: AnyStr, **kwargs):
        super().__init__(**kwargs)
        self.content = question


class ToolAction(BaseAction):
    def __init__(self, tool_name: AnyStr, tool_arg: Dict[AnyStr, Any], **kwargs):
        super().__init__(**kwargs)
        self.tool_name = tool_name
        self.tool_arg = tool_arg
        self.tool_result = ""


class LLMAction(BaseAction):
    def __init__(self, reply: AnyStr, **kwargs):
        super().__init__(**kwargs)
        self.reply = reply


class FinalAction(BaseAction):
    def __init__(self, return_val: AnyStr, **kwargs):
        super().__init__(**kwargs)
        self.return_val = return_val

nonebot_plugin_agent/action/base.py
Dobiichi-Origami/nonebot-plugin-agent
import asyncio
import datetime
import json
from typing import List, AnyStr, Callable, Optional, Any, Dict

from nonebot import logger

from .action import BaseAction, ToolAction, FinalAction, LLMAction, HumanAction
from .llm import Tool, LLM_Type, OpenAIFunctionCallLLM


def async_wrapper(func: Callable) -> Callable:
    async def new_async_wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return new_async_wrapper


def convert_result_into_str(result: Any) -> AnyStr:
    if isinstance(result, str):
        return result
    elif isinstance(result, bytes):
        return str(result, encoding="utf8")
    elif hasattr(result, "__str__"):
        return result.__str__
    else:
        return json.dumps(result, ensure_ascii=False)


class FunctionCaller:
    def __init__(
            self,
            llm_type: str = LLM_Type.OPENAI,
            max_calling_round: Optional[int] = 10,
            max_timeout_in_seconds: Optional[int] = 10,
            **kwargs
    ):
        if llm_type == LLM_Type.OPENAI:
            llm = OpenAIFunctionCallLLM(**kwargs)
        else:
            raise TypeError("不支持的 LLM 类型")

        self.function_call_llm = llm
        self.max_calling_round = max_calling_round
        self.max_timeout_in_seconds = max_timeout_in_seconds
        self.history_messages: Dict[AnyStr, List[BaseAction]] = {}

    async def process_with_tool(
            self,
            user_id: AnyStr,
            user_input: AnyStr,
            **kwargs,
    ) -> AnyStr:
        calling_round = 0
        start_timestamp = datetime.datetime.now()
        tools = Tool.get_tool_list()
        intermedia_actions = self.history_messages.get(user_id, [])
        intermedia_actions.append(HumanAction(question=user_input))

        try:
            while calling_round <= self.max_calling_round \
                    and (datetime.datetime.now() - start_timestamp).seconds <= self.max_timeout_in_seconds:
                next_action: BaseAction = await self.function_call_llm.async_generate_next_action(
                    tools=tools,
                    intermedia_actions=intermedia_actions,
                    inputs=user_input,
                )

                intermedia_actions.append(next_action)
                if isinstance(next_action, FinalAction):
                    return next_action.return_val
                elif isinstance(next_action, LLMAction):
                    return next_action.reply
                elif isinstance(next_action, ToolAction):
                    tool_need_for_calling = tools[next_action.tool_name].function
                    if asyncio.iscoroutinefunction(tool_need_for_calling):
                        func = tool_need_for_calling
                    else:
                        func = async_wrapper(tool_need_for_calling)
                    tool_result = await func(**next_action.tool_arg)
                    logger.info("工具 {} 输出: ".format(next_action.tool_name) + tool_result)
                    intermedia_actions[-1].tool_result = tool_result

                calling_round += 1

            return "等待超时或重试轮次过多"
        finally:
            logger.info("保存历史信息")
            self.history_messages[user_id] = intermedia_actions[-10:]

    async def __call__(self, *args, **kwargs):
        return await self.process_with_tool(*args, **kwargs)




nonebot_plugin_agent/base.py
Dobiichi-Origami/nonebot-plugin-agent
from typing import Dict, Optional, Any

from pydantic import BaseModel, Field
from .llm import LLM_Type


class AgentConfig(BaseModel):
    agent_llm_type: str = Field(default=LLM_Type.OPENAI)
    agent_llm_kwargs: Optional[Dict[str, Any]] = Field(default={})

    agent_max_calling_round_per_query: int = Field(default=10)
    agent_max_timeout_in_seconds_per_request: int = Field(default=10)

nonebot_plugin_agent/config.py
Dobiichi-Origami/nonebot-plugin-agent
from .base import BaseFunctionCallLLM, Tool, LLM_Type
from .openai import OpenAIFunctionCallLLM

__all__ = [
    "LLM_Type",
    "BaseFunctionCallLLM",
    "OpenAIFunctionCallLLM",
    "Tool",
]
nonebot_plugin_agent/llm/__init__.py
Dobiichi-Origami/nonebot-plugin-agent
import enum
from abc import ABC, abstractmethod
from copy import deepcopy

from typing import AnyStr, List, Callable, Dict, Any, Optional
from ..action import BaseAction


class Tool:

    _tool_maps: Dict = {}

    def __init__(
            self,
            function: Callable,
            param_list: Dict[AnyStr, Any],
            description: Optional[AnyStr],
    ):
        self.name = function.__name__
        self.function = function
        self.param_list = param_list
        self.description = description
        self.required = []

    @classmethod
    def add_to_tool_list(
        cls,
        function: Callable,
        param_list: Dict[AnyStr, Any],
        description: Optional[AnyStr],
    ):
        t = Tool(function, param_list, description)
        cls._tool_maps[t.name] = t

    @classmethod
    def get_tool_list(cls) -> Dict:
        return deepcopy(cls._tool_maps)


class LLM_Type:
    OPENAI = "openai"


class BaseFunctionCallLLM(ABC):
    @abstractmethod
    async def async_generate_next_action(
            self,
            tools: Dict[AnyStr, Tool],
            intermedia_actions: List[BaseAction],
            inputs: AnyStr
    ) -> BaseAction:
        ...



nonebot_plugin_agent/llm/base.py
Dobiichi-Origami/nonebot-plugin-agent
import json
import os

from openai import ChatCompletion
from nonebot import logger

from .base import BaseFunctionCallLLM, Tool
from typing import AnyStr, Optional, List, Any, Tuple, Dict
from ..action import BaseAction, ToolAction, HumanAction, LLMAction, FinalAction

_CHOICES = 'choices'
_FINISH_REASON = "finish_reason"
_MESSAGE = "message"
_FUNCTION_CALL = "function_call"

_STATUS_FUNCTION_CALL = "function_call"
_STATUS_STOP = "stop"


class OpenAIFunctionCallLLM(BaseFunctionCallLLM):
    def __init__(
            self,
            open_api_key: Optional[AnyStr] = None,
            open_api_base: Optional[AnyStr] = None,
            open_api_model: Optional[AnyStr] = "gpt-3.5-turbo-0613",
            **kwargs
    ):
        self.api_key = open_api_key if open_api_key else os.environ.get("OPENAI_API_KEY")
        self.api_base = open_api_base if open_api_base \
            else os.environ.get("OPENAI_API_BASE") if os.environ.get("OPENAI_API_BASE") else None
        self.api_model = open_api_model

    async def async_generate_next_action(
            self,
            tools: Dict[AnyStr, Tool],
            intermedia_actions: List[BaseAction],
            inputs: AnyStr
    ) -> BaseAction:
        tool_list = self.__class__._get_tool_list(tools)
        history_messages = self.__class__._get_message(intermedia_actions)

        logger.info("发送信息:" + json.dumps(history_messages, ensure_ascii=False))
        response = await self._get_chat_completion(messages=history_messages, functions=tool_list)
        logger.info("OpenAI 回复: " + json.dumps(response, ensure_ascii=False))

        message = response[_CHOICES][0]
        status = message[_FINISH_REASON]
        result = message[_MESSAGE]["content"]

        if status == _STATUS_FUNCTION_CALL:
            call_info = message[_MESSAGE][_FUNCTION_CALL]
            if len(call_info["arguments"]) != 0:
                return ToolAction(tool_name=call_info["name"], tool_arg=json.loads(call_info["arguments"]))
            else:
                return LLMAction(reply=result)
        elif status == _STATUS_STOP:
            if await self._check_is_problem_solved(inputs, result):
                return FinalAction(return_val=result)
            else:
                return LLMAction(reply=result)
        else:
            return FinalAction(return_val="error")

    async def _check_is_problem_solved(self, inputs: AnyStr, answer: AnyStr) -> bool:
        messages = [
            {
                "role": "system",
                "content": """
                your duty is to check whether user's question has benn answered.
                If question has been answered, you should and only should answer "YES", no spare word.
                Similarly, answer "NO" only if the question remains unresolved.
                """,
            },
            {
                "role": "user",
                "content": """
                there is a conversation below, you need to tell whether the question has beem answered.
                
                question: {inputs}
                answer: {answer},
                
                Next, give your judgement:
                """.format(inputs=inputs, answer=answer),
            },
        ]

        message: str = (await self._get_chat_completion(messages=messages))[_CHOICES][0][_MESSAGE]["content"]
        logger.info("检查信息发送: " + message)
        return message.upper() == "YES"

    async def _get_chat_completion(self, **kwargs):
        return await ChatCompletion.acreate(
            api_key=self.api_key,
            api_base=self.api_base,
            model=self.api_model,
            **kwargs
        )

    @staticmethod
    def _get_tool_list(tools: Dict[AnyStr, Tool]) -> List[Dict[AnyStr, Any]]:
        result: List[Dict[AnyStr, Any]] = []
        for tool_name, tool in tools.items():
            result.append({
                "name": tool_name,
                "description": tool.description,
                "parameters": {
                    "type": "object",
                    "properties": tool.param_list
                },
                "required": tool.required
            })
        return result

    @staticmethod
    def _get_message(
            intermedia_actions: List[BaseAction],
            prompt: Optional[AnyStr] = None
    ) -> List[Dict[AnyStr, AnyStr]]:
        prompt = prompt if prompt else "you are a helpful assistant, you need to use function to solve people' question"
        messages = [{"role": "system", "content": prompt}]
        for action in intermedia_actions:
            if isinstance(action, ToolAction):
                messages.append({
                    "role": "function",
                    "name": action.tool_name,
                    "content": action.tool_result,
                })
            elif isinstance(action, HumanAction):
                messages.append({
                    "role": "user",
                    "content": action.content,
                })
            elif isinstance(action, LLMAction):
                messages.append({
                    "role": "assistant",
                    "content": action.reply
                })
            elif isinstance(action, FinalAction):
                messages.append({
                    "role": "assistant",
                    "content": action.return_val
                })
            else:
                raise TypeError("未知的 Action 类型: " + str(type(action)))
        return messages

nonebot_plugin_agent/llm/openai.py
Dobiichi-Origami/nonebot-plugin-agent