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