agentUniverse/docs/guidebook/zh/In-Depth_Guides/原理介绍/计划/计划定义与使用.md

如何定义计划Planner组件

根据agentUniverse领域组件的设计特性，同其他组件一样，创建一个计划Planner定义由2部分组成:

• planner_xx.yaml
• planner_xx.py

其中planner_xx.yaml包含了Planner组件的名称、描述、输入、输出等重要信息；planner_xx.py包含了的Planner的具体行为定义。理解这一原理后，让我们具体看看该如何创建这两部分内容。

创建Planner配置 - planner_xx.yaml

我们将详细介绍配置中的各组成部分。

设置Planner的基本属性

• name: Planner名称，您可以按照自己的期望设置任何名字,唯一标识
• description: planner的描述说明
• input_key: 默认为input，实际输入对应的key
• output_key: 默认为output，实际输出对应的key

设置Planner组件元信息

metadata - 组件元信息

• type : 组件类型，'PLANNER'
• module: Planner实体包路径
• class: Planner实体类名

一个Planner定义配置的实际样例

name: 'expressing_planner'
description: 'expressing planner'
metadata:
  type: 'PLANNER'
  module: 'agentuniverse.agent.plan.planner.expressing_planner.expressing_planner'
  class: 'ExpressingPlanner'

上述是一个实际的Planner配置的样例。除了上述介绍的标准配置项。

您可以在工程中的agentuniverse.agent.plan.planner路径下查看更多的planner配置yaml样例。

除此之外agentuniverse不限制用户对Planner yaml配置内容进行扩展，您可以根据自己的要求创建任何自定义配置key,但请注意不要与上述默认的配置关键字重名。

创建Planner领域行为定义 - planner_xx.py

创建Planner类对象

创建对应的Planner类对象并继承agentUniverse框架Planner基础类 Planner;

编写Planner类对象invoke方法

在invoke方法中编写计划的实际逻辑片段。

def invoke(self, agent_model: AgentModel, planner_input: dict, input_object: InputObject) -> dict:
    """Invoke the planner.

    Args:
        agent_model (AgentModel): Agent model object.
        planner_input (dict): Planner input object.
        input_object (InputObject): The input parameters passed by the user.
    Returns:
        dict: The planner result.
    """
    planner_config = agent_model.plan.get('planner')
    sub_agents = self.generate_sub_agents(planner_config)
    return self.agents_run(sub_agents, planner_config, planner_input, input_object)

上述案例以peer planner作为说明，其选择了peer中的各个子agent，并按照peer的模式进行运行。在peer的说明部分我们会进一步详细介绍peer的工作原理，这里我们不展开说明。

同理，下面的非必须方法我们在此也不详细介绍了，建议您以一个实际的planner源码作为参考结合观看，例如'agentuniverse.agent.plan.planner.planning_planner.planning_planner'.

编写handle_prompt方法

在handle_prompt方法中编写planner中关于prompt处理的方法，非必须。

编写handle_memory方法

在handle_memory方法中编写planner中关于memory处理的方法，非必须。

编写handle_all_actions方法

在handle_all_actions方法中编写planner中关于所有action的处理的方法，非必须。

编写handle_llm方法

在handle_llm方法中编写planner中关于llm处理的方法，非必须。

一个实际的planner对象定义样例

import asyncio

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

from agentuniverse.agent.agent_model import AgentModel
from agentuniverse.agent.input_object import InputObject
from agentuniverse.agent.memory.chat_memory import ChatMemory
from agentuniverse.agent.plan.planner.planner import Planner
from agentuniverse.base.util.memory_util import generate_memories
from agentuniverse.base.util.prompt_util import process_llm_token
from agentuniverse.llm.llm import LLM
from agentuniverse.prompt.chat_prompt import ChatPrompt
from agentuniverse.prompt.prompt import Prompt
from agentuniverse.prompt.prompt_manager import PromptManager
from agentuniverse.prompt.prompt_model import AgentPromptModel


class RagPlanner(Planner):
    """Rag planner class."""

    def invoke(self, agent_model: AgentModel, planner_input: dict,
               input_object: InputObject) -> dict:
        """Invoke the planner.

        Args:
            agent_model (AgentModel): Agent model object.
            planner_input (dict): Planner input object.
            input_object (InputObject): The input parameters passed by the user.
        Returns:
            dict: The planner result.
        """
        memory: ChatMemory = self.handle_memory(agent_model, planner_input)

        self.run_all_actions(agent_model, planner_input, input_object)

        llm: LLM = self.handle_llm(agent_model)

        prompt: ChatPrompt = self.handle_prompt(agent_model, planner_input)
        process_llm_token(llm, prompt.as_langchain(), agent_model.profile, planner_input)

        chat_history = memory.as_langchain().chat_memory if memory else InMemoryChatMessageHistory()

        chain_with_history = RunnableWithMessageHistory(
            prompt.as_langchain() | llm.as_langchain(),
            lambda session_id: chat_history,
            history_messages_key="chat_history",
            input_messages_key=self.input_key,
        )
        res = asyncio.run(
            chain_with_history.ainvoke(input=planner_input, config={"configurable": {"session_id": "unused"}}))
        return {**planner_input, self.output_key: res.content}

    def handle_prompt(self, agent_model: AgentModel, planner_input: dict) -> ChatPrompt:
        """Prompt module processing.

        Args:
            agent_model (AgentModel): Agent model object.
            planner_input (dict): Planner input object.
        Returns:
            ChatPrompt: The chat prompt instance.
        """
        profile: dict = agent_model.profile

        profile_prompt_model: AgentPromptModel = AgentPromptModel(introduction=profile.get('introduction'),
                                                                  target=profile.get('target'),
                                                                  instruction=profile.get('instruction'))

        # get the prompt by the prompt version
        prompt_version: str = profile.get('prompt_version')
        version_prompt: Prompt = PromptManager().get_instance_obj(prompt_version)

        if version_prompt is None and not profile_prompt_model:
            raise Exception("Either the `prompt_version` or `introduction & target & instruction`"
                            " in agent profile configuration should be provided.")
        if version_prompt:
            version_prompt_model: AgentPromptModel = AgentPromptModel(
                introduction=getattr(version_prompt, 'introduction', ''),
                target=getattr(version_prompt, 'target', ''),
                instruction=getattr(version_prompt, 'instruction', ''))
            profile_prompt_model = profile_prompt_model + version_prompt_model

        chat_prompt = ChatPrompt().build_prompt(profile_prompt_model, self.prompt_assemble_order)
        image_urls: list = planner_input.pop('image_urls', []) or []
        if image_urls:
            chat_prompt.generate_image_prompt(image_urls)
        return chat_prompt

上述是一个基于Rag planner的实际样例，我们在此不展开说明，可以关注rag planner介绍部分。

关注您定义的Planner所在的包路径

通过上面的Planner配置与定义，您已经掌握了Planner创建的所有步骤；接下去我们将使用这些Planner，在使用前请关注创建的Planner是否在正确的包扫描路径内。

在agentUniverse项目的config.toml中需要配置Planner配置对应的package, 请再次确认您创建的文件所在的包路径是否在CORE_PACKAGE中planner路径或其子路径下。

以示例工程中的配置为例，如下：

[CORE_PACKAGE]
# Scan and register planner components for all paths under this list, with priority over the default.
planner = ['sample_standard_app.app.core.planner']

如何使用计划Planner组件

在Agent中配置使用

您可以根据智能体创建与使用中的内容在agent的planner中设置您创建的任意计划组件。

使用Planner管理器

通过Planner管理器中的.get_instance_obj(xx_planner_name) 方法可以获取对应名称的Planner实例, 通过invoke方法进行调用。

from agentuniverse.agent.plan.planner.planner_manager import PlannerManager

planner_base = PlannerManager().get_instance_obj('xx_planner_name')
planner_result = planner_base.invoke(self.agent_model, agent_input, input_object)

了解更多已有计划Planner组件

框架提供的更多Planner组件在agentuniverse.agent.plan.planner包路径下，您可以进一步查看对应代码或在我们的扩展组件介绍部分进一步了解他们，更多已经接入的Planner您可以在Planner列表章节中了解。

总结

至此您已经掌握了计划Planner的定义与使用，赶快去尝试定义与使用Planner吧。