layer_operation_plan.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162
  1. import json
  2. from qwen_agent.planning.planner import Planner
  3. from qwen_agent.memory.plan_memory import PlanExampleRetrieval
  4. actions_list = {
  5. "GisLayerOperationAgent": "用户控制图层的打开和关闭",
  6. "summary": "对查询的结果,进行总结摘要,提炼用户关注的信息"
  7. }
  8. PROMPT_TEMPLATE = """
  9. 你是一个GIS图层控制系统,需要将自然语言指令解析为结构化操作
  10. 如果用户历史回答中已经相应的答案,请直接使用summary对历史回答进行总结
  11. 下面是动作选项列表,可以选择有助于完成用户需求的一个或多个动作:
  12. 动作列表
  13. {actions_list}
  14. 请依据参考资料,制定计划完成用户需求,按照如下格式返回:
  15. Question: 用户针区域土地利用现状问题的提问
  16. Thought: 生成Plan的思考过程,如果提到多个区域名称,需要多次调用[GisLayerOperationAgent]接口,给用户一些依据。
  17. Plan: 生成的计划,包含函数名和执行的目标。以JSON格式返回,所有的Action会以执行的先后顺序保存在list中,例如:
  18. [{{"action_name": 第一步执行的Action名字, "instruction": 执行需要达到的预期,请给出详细有效的指示}}, {{"action_name": 第二步执行的Action名字, "instruction": 执行需要达到的预期,请给出详细有效的指示}}]
  19. 举例:
  20. {example_list}
  21. 注意0:最后一步必须调用[summary]这个action,对查询的结果,进行总结摘要,提炼用户关注的信息;
  22. """
  23. examples = """
  24. Examples #1:
  25. Question: 打开永久基本农田图层
  26. Thought: 用户问题中想要打开永久基本农田图层,所以需要通过[GisLayerOperationAgent]将问题转换为具体的结构化指令
  27. Plan: ```json\n [{\"action_name\": \"GisLayerOperationAgent\", \"instruction\": \"你需要调用 [GisLayerOperationAgent]将问题转换为具体的结构化指令\"}, {\"action_name\": \"summary\",\"instruction\": \"对查询的结果,进行总结摘要,提炼用户关注的信息\"}]
  28. ```
  29. """
  30. DEFAULT_PLAN = """使用默认执行计划:
  31. Plan: ```json\n [{\"action_name\": \"GisLayerOperationAgent\", \"instruction\": \"你需要调用 [GisLayerOperationAgent]将问题转换为具体的结构化指令\"}]
  32. """
  33. class LayerOperationPlan(Planner):
  34. def __init__(self, llm_name, name='layer_operation', **kvargs):
  35. self.actions_list_str = json.dumps(actions_list, ensure_ascii=False)
  36. # retriever = PlanExampleRetrieval(query_type=name)
  37. super(LayerOperationPlan, self).__init__(
  38. llm_name=llm_name,
  39. name=f'{name}_planner',
  40. system_prompt=PROMPT_TEMPLATE,
  41. retriever=None,
  42. default_plan=DEFAULT_PLAN, **kvargs
  43. )
  44. def get_system_prompt(self, user_request):
  45. # similar_exams = self.retriever.get_relevant_documents(user_request, top_k=3)
  46. # examples = '\n'.join([f"Example #{i + 1}:\n{q}" for i, (n, q) in enumerate(similar_exams[::-1])])
  47. # print(f'user request:{user_request} \nRetrivaled Examples: {examples}\n')
  48. return PROMPT_TEMPLATE.format(actions_list=self.actions_list_str, example_list=None)