CSPON-TSC/aiAgent_gd/qwen_agent/messages/plan_message.py

from pydantic import BaseModel, Field
from typing import List


class PlanInfo(BaseModel):
    action_name: str
    instruction: str
    # output: str = None
    # enable: bool = True
    executed: bool = False
    # def __str__(self):
    #     if self.executed:
    #         if self.idx != -1:
    #             return f"{self.idx}. 通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
    #             # return f"{self.idx}. 通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
    #         else:
    #             return f"通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
    #     else:
    #         return f"Instruction:{self.instruction}"


class PlanExecuteContextManager(BaseModel):
    finish_agent_list: List[PlanInfo] = []
    execute_agent: PlanInfo = None
    user_request: str
    plan_msg: str = None

    def append(self, plan: PlanInfo):
        plan.idx = len(self.finish_list) + 1
        self.finish_list.append(plan)

    def pop(self):
        return self.finish_list.pop()
    def set_last_plan_execute(self, output=''):
        assert self.execute_agent is not None
        self.execute_agent.executed = True
        self.execute_agent.output = output
        self.finish_agent_list.append(self.execute_agent)
        self.execute_agent = None
        # self.finish_list[-1].executed = True
        # self.finish_list[-1].output = output

    def get_last_plan_output(self):
        return self.finish_list[-1].output

    def get_context(self, add_user_request=True, executed=True, add_plan_msg=True):
        result = '' 
        if self.user_request and add_user_request:
            result += f'用户的原始Question为：{self.user_request}\n'

        if self.plan_msg and add_plan_msg:
            result += self.plan_msg

        if executed == True:
            plan_str = '\n'.join([str(plan) for plan in self.finish_list if plan.enable])
            result += f"已经执行结束的Plan如下：\n{plan_str}\n"
        else:
            plan_str = '\n'.join([str(plan) for plan in self.finish_list if plan.enable and plan.executed == False])
            result += f"需要执行的Plan如下：\n{plan_str}\n"

        return result