DB-GPT扩展自定义Agent配置说明

打印 上一主题 下一主题

主题 2046|帖子 2046|积分 6138

简介

文章主要介绍了怎样扩展一个自定义Agent,这里是用官方提供的总结摘要的Agent做了个示例,先给大家看下表现结果


代码目次

博主将代码放在core目次了,后续经过对源码的解读感觉放在dbgpt_serve.agent.agents.expand目次下大概更合适,大家自行把控即可

代码详情

summarizer_action.py

from typing import Optional
from pydantic import BaseModel, Field
from dbgpt.vis import Vis
from dbgpt.agent import Action, ActionOutput, AgentResource, ResourceType
from dbgpt.agent.util import cmp_string_equal

NOT_RELATED_MESSAGE = "Did not find the information you want."


# The parameter object that the Action that the current Agent needs to execute needs to output.
class SummaryActionInput(BaseModel):
    summary: str = Field(
        ...,
        description="The summary content",
    )


class SummaryAction(Action[SummaryActionInput]):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    @property
    def resource_need(self) -> Optional[ResourceType]:
        # The resource type that the current Agent needs to use
        # here we do not need to use resources, just return None
        return None

    @property
    def render_protocol(self) -> Optional[Vis]:
        # The visualization rendering protocol that the current Agent needs to use
        # here we do not need to use visualization rendering, just return None
        return None

    @property
    def out_model_type(self):
        return SummaryActionInput

    async def run(
            self,
            ai_message: str,
            resource: Optional[AgentResource] = None,
            rely_action_out: Optional[ActionOutput] = None,
            need_vis_render: bool = True,
            **kwargs,
    ) -> ActionOutput:
        """erform the action.

        The entry point for actual execution of Action. Action execution will be
        automatically initiated after model inference.
        """
        try:
            # Parse the input message
            param: SummaryActionInput = self._input_convert(ai_message, SummaryActionInput)
        except Exception:
            return ActionOutput(
                is_exe_success=False,
                content="The requested correctly structured answer could not be found, "
                        f"ai message: {ai_message}",
            )
        # Check if the summary content is not related to user questions
        if param.summary and cmp_string_equal(
                param.summary,
                NOT_RELATED_MESSAGE,
                ignore_case=True,
                ignore_punctuation=True,
                ignore_whitespace=True,
        ):
            return ActionOutput(
                is_exe_success=False,
                content="the provided text content is not related to user questions at all."
                        f"ai message: {ai_message}",
            )
        else:
            return ActionOutput(
                is_exe_success=True,
                content=param.summary,
            )

summarizer_agent.py

from typing import Optional
from pydantic import BaseModel, Field
from dbgpt.vis import Vis
from dbgpt.agent import Action, ActionOutput, AgentResource, ResourceType
from dbgpt.agent.util import cmp_string_equal

NOT_RELATED_MESSAGE = "Did not find the information you want."


# The parameter object that the Action that the current Agent needs to execute needs to output.
class SummaryActionInput(BaseModel):
    summary: str = Field(
        ...,
        description="The summary content",
    )


class SummaryAction(Action[SummaryActionInput]):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    @property
    def resource_need(self) -> Optional[ResourceType]:
        # The resource type that the current Agent needs to use
        # here we do not need to use resources, just return None
        return None

    @property
    def render_protocol(self) -> Optional[Vis]:
        # The visualization rendering protocol that the current Agent needs to use
        # here we do not need to use visualization rendering, just return None
        return None

    @property
    def out_model_type(self):
        return SummaryActionInput

    async def run(
            self,
            ai_message: str,
            resource: Optional[AgentResource] = None,
            rely_action_out: Optional[ActionOutput] = None,
            need_vis_render: bool = True,
            **kwargs,
    ) -> ActionOutput:
        """erform the action.

        The entry point for actual execution of Action. Action execution will be
        automatically initiated after model inference.
        """
        try:
            # Parse the input message
            param: SummaryActionInput = self._input_convert(ai_message, SummaryActionInput)
        except Exception:
            return ActionOutput(
                is_exe_success=False,
                content="The requested correctly structured answer could not be found, "
                        f"ai message: {ai_message}",
            )
        # Check if the summary content is not related to user questions
        if param.summary and cmp_string_equal(
                param.summary,
                NOT_RELATED_MESSAGE,
                ignore_case=True,
                ignore_punctuation=True,
                ignore_whitespace=True,
        ):
            return ActionOutput(
                is_exe_success=False,
                content="the provided text content is not related to user questions at all."
                        f"ai message: {ai_message}",
            )
        else:
            return ActionOutput(
                is_exe_success=True,
                content=param.summary,
            )

如许重启项目就能看到自定义的agent了


免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有账号?立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

麻花痒

论坛元老
这个人很懒什么都没写!
快速回复 返回顶部 返回列表