plan_message.py 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. from pydantic import BaseModel, Field
  2. from typing import List
  3. class PlanInfo(BaseModel):
  4. action_name: str
  5. instruction: str
  6. # output: str = None
  7. # enable: bool = True
  8. executed: bool = False
  9. # def __str__(self):
  10. # if self.executed:
  11. # if self.idx != -1:
  12. # return f"{self.idx}. 通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
  13. # # return f"{self.idx}. 通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
  14. # else:
  15. # return f"通过执行[{self.action_name}]接口\nInstruction:{self.instruction}\nOutput:{self.output}"
  16. # else:
  17. # return f"Instruction:{self.instruction}"
  18. class PlanExecuteContextManager(BaseModel):
  19. finish_agent_list: List[PlanInfo] = []
  20. execute_agent: PlanInfo = None
  21. user_request: str
  22. plan_msg: str = None
  23. def append(self, plan: PlanInfo):
  24. plan.idx = len(self.finish_list) + 1
  25. self.finish_list.append(plan)
  26. def pop(self):
  27. return self.finish_list.pop()
  28. def set_last_plan_execute(self, output=''):
  29. assert self.execute_agent is not None
  30. self.execute_agent.executed = True
  31. self.execute_agent.output = output
  32. self.finish_agent_list.append(self.execute_agent)
  33. self.execute_agent = None
  34. # self.finish_list[-1].executed = True
  35. # self.finish_list[-1].output = output
  36. def get_last_plan_output(self):
  37. return self.finish_list[-1].output
  38. def get_context(self, add_user_request=True, executed=True, add_plan_msg=True):
  39. result = ''
  40. if self.user_request and add_user_request:
  41. result += f'用户的原始Question为:{self.user_request}\n'
  42. if self.plan_msg and add_plan_msg:
  43. result += self.plan_msg
  44. if executed == True:
  45. plan_str = '\n'.join([str(plan) for plan in self.finish_list if plan.enable])
  46. result += f"已经执行结束的Plan如下:\n{plan_str}\n"
  47. else:
  48. plan_str = '\n'.join([str(plan) for plan in self.finish_list if plan.enable and plan.executed == False])
  49. result += f"需要执行的Plan如下:\n{plan_str}\n"
  50. return result