From 8678e3689920a7e284157e7a5290fe07bcd1a599 Mon Sep 17 00:00:00 2001 From: z30057876 Date: Thu, 30 Oct 2025 15:34:32 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=A7=E7=BB=AD=E4=BC=98=E5=8C=96Prompt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/scheduler/call/facts/facts.py | 16 +- apps/scheduler/call/facts/prompt.py | 107 ++++-- apps/scheduler/call/rag/prompt.py | 52 ++- apps/scheduler/call/rag/rag.py | 3 +- apps/scheduler/call/suggest/prompt.py | 97 +++-- apps/scheduler/call/suggest/suggest.py | 4 +- apps/scheduler/mcp/host.py | 63 +++- apps/scheduler/mcp/plan.py | 50 +-- apps/scheduler/mcp/prompt.py | 502 +++++++++++++++---------- 9 files changed, 545 insertions(+), 349 deletions(-) diff --git a/apps/scheduler/call/facts/facts.py b/apps/scheduler/call/facts/facts.py index 5f2bffbec..5401531ab 100644 --- a/apps/scheduler/call/facts/facts.py +++ b/apps/scheduler/call/facts/facts.py @@ -13,7 +13,7 @@ from apps.schemas.enum_var import CallOutputType from apps.schemas.scheduler import CallInfo, CallOutputChunk, CallVars from apps.services.user_tag import UserTagManager -from .prompt import DOMAIN_PROMPT, FACTS_PROMPT +from .prompt import DOMAIN_FUNCTION, DOMAIN_PROMPT, FACTS_FUNCTION, FACTS_PROMPT from .schema import ( DomainGen, FactsGen, @@ -80,17 +80,14 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): # 组装conversation消息 facts_prompt = FACTS_PROMPT[self._sys_vars.language] facts_conversation = [ + {"role": "system", "content": "You are a helpful assistant."}, *data.message, {"role": "user", "content": facts_prompt}, ] # 提取事实信息 facts_result = await json_generator.generate( - function={ - "name": "extract_facts", - "description": "Extract facts from the conversation", - "parameters": FactsGen.model_json_schema(), - }, + function=FACTS_FUNCTION[self._sys_vars.language], conversation=facts_conversation, language=self._sys_vars.language, ) @@ -99,17 +96,14 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): # 组装conversation消息 domain_prompt = DOMAIN_PROMPT[self._sys_vars.language] domain_conversation = [ + {"role": "system", "content": "You are a helpful assistant."}, *data.message, {"role": "user", "content": domain_prompt}, ] # 更新用户画像 domain_result = await json_generator.generate( - function={ - "name": "extract_domain", - "description": "Extract domain keywords from the conversation", - "parameters": DomainGen.model_json_schema(), - }, + function=DOMAIN_FUNCTION[self._sys_vars.language], conversation=domain_conversation, language=self._sys_vars.language, ) diff --git a/apps/scheduler/call/facts/prompt.py b/apps/scheduler/call/facts/prompt.py index e4cfb9b7f..22fab49f8 100644 --- a/apps/scheduler/call/facts/prompt.py +++ b/apps/scheduler/call/facts/prompt.py @@ -137,42 +137,89 @@ include, parallel, mutually exclusive, etc. ), } -DOMAIN_FUNCTION: dict[str, Any] = { - "name": "extract_domain", - "description": "从对话中提取领域关键词标签 / Extract domain keyword tags from conversation", - "parameters": { - "type": "object", - "properties": { - "keywords": { - "type": "array", - "items": {"type": "string"}, - "description": "关键词或标签列表 / List of keywords or tags", +DOMAIN_FUNCTION: dict[LanguageType, dict[str, Any]] = { + LanguageType.CHINESE: { + "name": "extract_domain", + "description": "从对话中提取领域关键词标签", + "parameters": { + "type": "object", + "properties": { + "keywords": { + "type": "array", + "items": {"type": "string"}, + "description": "关键词或标签列表", + }, }, + "required": ["keywords"], }, - "required": ["keywords"], + "examples": [ + {"keywords": ["北京", "天气"]}, + {"keywords": ["Python", "装饰器", "设计模式"]}, + ], + }, + LanguageType.ENGLISH: { + "name": "extract_domain", + "description": "Extract domain keyword tags from conversation", + "parameters": { + "type": "object", + "properties": { + "keywords": { + "type": "array", + "items": {"type": "string"}, + "description": "List of keywords or tags", + }, + }, + "required": ["keywords"], + }, + "examples": [ + {"keywords": ["Beijing", "weather"]}, + {"keywords": ["Python", "decorator", "design pattern"]}, + ], }, - "examples": [ - {"keywords": ["北京", "天气"]}, - {"keywords": ["Python", "装饰器", "设计模式"]}, - ], } -FACTS_FUNCTION: dict[str, Any] = { - "name": "extract_facts", - "description": "从对话中提取关键事实信息 / Extract key fact information from conversation", - "parameters": { - "type": "object", - "properties": { - "facts": { - "type": "array", - "items": {"type": "string"}, - "description": "从对话中提取的事实条目 / Fact entries extracted from conversation", +FACTS_FUNCTION: dict[LanguageType, dict[str, Any]] = { + LanguageType.CHINESE: { + "name": "extract_facts", + "description": "从对话中提取关键事实信息", + "parameters": { + "type": "object", + "properties": { + "facts": { + "type": "array", + "items": {"type": "string"}, + "description": "从对话中提取的事实条目", + }, }, + "required": ["facts"], }, - "required": ["facts"], + "examples": [ + {"facts": ["杭州西湖有苏堤、白堤、断桥、三潭印月等景点"]}, + {"facts": ["用户喜欢看科幻电影", "用户可能对《星际穿越》感兴趣"]}, + ], + }, + LanguageType.ENGLISH: { + "name": "extract_facts", + "description": "Extract key fact information from conversation", + "parameters": { + "type": "object", + "properties": { + "facts": { + "type": "array", + "items": {"type": "string"}, + "description": "Fact entries extracted from conversation", + }, + }, + "required": ["facts"], + }, + "examples": [ + { + "facts": [ + "Hangzhou West Lake has Su Causeway, Bai Causeway, Broken Bridge, " + "Three Pools Mirroring the Moon, etc.", + ], + }, + {"facts": ["User likes watching sci-fi movies", "User may be interested in Interstellar"]}, + ], }, - "examples": [ - {"facts": ["杭州西湖有苏堤、白堤、断桥、三潭印月等景点"]}, - {"facts": ["用户喜欢看科幻电影", "用户可能对《星际穿越》感兴趣"]}, - ], } diff --git a/apps/scheduler/call/rag/prompt.py b/apps/scheduler/call/rag/prompt.py index b4d849d2f..cd6560fe6 100644 --- a/apps/scheduler/call/rag/prompt.py +++ b/apps/scheduler/call/rag/prompt.py @@ -68,27 +68,39 @@ abbreviations, etc.) ).strip(), } -QUESTION_REWRITE_FUNCTION: dict[str, object] = { - "name": "rewrite_question", - "description": ( - "基于上下文优化用户问题，使其更适合知识库检索 / " - "Optimize user question based on context for better knowledge base retrieval" - ), - "parameters": { - "type": "object", - "properties": { - "question": { - "type": "string", - "description": ( - "优化后的问题。应该完整、明确、包含关键信息，便于知识库检索 / " - "The optimized question that is complete, clear, and retrieval-friendly" - ), +QUESTION_REWRITE_FUNCTION: dict[LanguageType, dict[str, object]] = { + LanguageType.CHINESE: { + "name": "rewrite_question", + "description": "基于上下文优化用户问题，使其更适合知识库检索", + "parameters": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "优化后的问题。应该完整、明确、包含关键信息，便于知识库检索", + }, }, + "required": ["question"], }, - "required": ["question"], + "examples": [ + {"question": "openEuler操作系统的优势和特点是什么？"}, + ], + }, + LanguageType.ENGLISH: { + "name": "rewrite_question", + "description": "Optimize user question based on context for better knowledge base retrieval", + "parameters": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The optimized question that is complete, clear, and retrieval-friendly", + }, + }, + "required": ["question"], + }, + "examples": [ + {"question": "How to install and configure Docker container engine on Linux system?"}, + ], }, - "examples": [ - {"question": "openEuler操作系统的优势和特点是什么？"}, - {"question": "How to install and configure Docker container engine on Linux system?"}, - ], } diff --git a/apps/scheduler/call/rag/rag.py b/apps/scheduler/call/rag/rag.py index 59c0eadbe..4058a3e2b 100644 --- a/apps/scheduler/call/rag/rag.py +++ b/apps/scheduler/call/rag/rag.py @@ -169,8 +169,9 @@ class RAG(CoreCall, input_model=RAGInput, output_model=RAGOutput): # 使用json_generator直接获取JSON结果 json_result = await json_generator.generate( - function=QUESTION_REWRITE_FUNCTION, + function=QUESTION_REWRITE_FUNCTION[self._sys_vars.language], conversation=[ + {"role": "system", "content": "You are a helpful assistant."}, *self._sys_vars.background.conversation[-self.history_len:], {"role": "user", "content": prompt}, ], diff --git a/apps/scheduler/call/suggest/prompt.py b/apps/scheduler/call/suggest/prompt.py index e90985e0d..3cbf9b0a5 100644 --- a/apps/scheduler/call/suggest/prompt.py +++ b/apps/scheduler/call/suggest/prompt.py @@ -5,43 +5,76 @@ from textwrap import dedent from apps.models import LanguageType -# Function Schema for question suggestion -SUGGEST_FUNCTION_SCHEMA = { - "name": "generate_suggestions", - "description": "Generate recommended follow-up questions based on conversation context and user interests / " - "基于对话上下文和用户兴趣生成推荐的后续问题", - "parameters": { - "type": "object", - "properties": { - "predicted_questions": { - "type": "array", - "description": "List of predicted questions, each should be a complete interrogative or imperative " - "sentence / 预测的问题列表，每个问题应该是完整的疑问句或祈使句", - "items": { - "type": "string", - "description": "Single recommended question, not exceeding 30 words / 单个推荐问题，长度不超过30字", +SUGGEST_FUNCTION: dict[LanguageType, dict] = { + LanguageType.CHINESE: { + "name": "generate_suggestions", + "description": "基于对话上下文和用户兴趣生成推荐的后续问题", + "parameters": { + "type": "object", + "properties": { + "predicted_questions": { + "type": "array", + "description": "预测的问题列表,每个问题应该是完整的疑问句或祈使句", + "items": { + "type": "string", + "description": "单个推荐问题,长度不超过30字", + }, }, }, + "required": ["predicted_questions"], }, - "required": ["predicted_questions"], + "examples": [ + { + "predicted_questions": [ + "杭州的最佳旅游季节是什么时候?", + "灵隐寺的开放时间和门票信息?", + "杭州有哪些适合亲子游的景点?", + ], + }, + { + "predicted_questions": [ + "字典和集合有什么特点?", + "如何在Python中处理异常?", + "列表推导式怎么使用?", + ], + }, + ], }, - "examples": [ - { - "predicted_questions": [ - "What is the best season to visit Hangzhou? / 杭州的最佳旅游季节是什么时候?", - "What are the opening hours and ticket information for Lingyin Temple? / " - "灵隐寺的开放时间和门票信息?", - "Which attractions in Hangzhou are suitable for family trips? / 杭州有哪些适合亲子游的景点?", - ], - }, - { - "predicted_questions": [ - "What are the characteristics of dictionaries and sets? / 字典和集合有什么特点?", - "How to handle exceptions in Python? / 如何在Python中处理异常?", - "How to use list comprehensions? / 列表推导式怎么使用?", - ], + LanguageType.ENGLISH: { + "name": "generate_suggestions", + "description": "Generate recommended follow-up questions based on conversation context and user interests", + "parameters": { + "type": "object", + "properties": { + "predicted_questions": { + "type": "array", + "description": "List of predicted questions, each should be a complete interrogative or " + "imperative sentence", + "items": { + "type": "string", + "description": "Single recommended question, not exceeding 30 words", + }, + }, + }, + "required": ["predicted_questions"], }, - ], + "examples": [ + { + "predicted_questions": [ + "What is the best season to visit Hangzhou?", + "What are the opening hours and ticket information for Lingyin Temple?", + "Which attractions in Hangzhou are suitable for family trips?", + ], + }, + { + "predicted_questions": [ + "What are the characteristics of dictionaries and sets?", + "How to handle exceptions in Python?", + "How to use list comprehensions?", + ], + }, + ], + }, } SUGGEST_PROMPT: dict[LanguageType, str] = { diff --git a/apps/scheduler/call/suggest/suggest.py b/apps/scheduler/call/suggest/suggest.py index 16505d500..143569f64 100644 --- a/apps/scheduler/call/suggest/suggest.py +++ b/apps/scheduler/call/suggest/suggest.py @@ -23,7 +23,7 @@ from apps.schemas.scheduler import ( ) from apps.services.user_tag import UserTagManager -from .prompt import SUGGEST_FUNCTION_SCHEMA, SUGGEST_PROMPT +from .prompt import SUGGEST_FUNCTION, SUGGEST_PROMPT from .schema import ( SingleFlowSuggestionConfig, SuggestGenResult, @@ -152,7 +152,7 @@ class Suggestion(CoreCall, input_model=SuggestionInput, output_model=SuggestionO {"role": "user", "content": prompt}, ] result = await json_generator.generate( - function=SUGGEST_FUNCTION_SCHEMA, + function=SUGGEST_FUNCTION[self._sys_vars.language], conversation=messages, language=self._sys_vars.language, ) diff --git a/apps/scheduler/mcp/host.py b/apps/scheduler/mcp/host.py index 33285ebc4..8ae8b5051 100644 --- a/apps/scheduler/mcp/host.py +++ b/apps/scheduler/mcp/host.py @@ -12,7 +12,7 @@ from mcp.types import TextContent from apps.llm import LLM, json_generator from apps.models import LanguageType, MCPTools -from apps.scheduler.mcp.prompt import MEMORY_TEMPLATE +from apps.scheduler.mcp.prompt import FILL_PARAMS_QUERY, MEMORY_TEMPLATE from apps.scheduler.pool.mcp.client import MCPClient from apps.scheduler.pool.mcp.pool import mcp_pool from apps.schemas.mcp import MCPContext, MCPPlanItem @@ -56,14 +56,42 @@ class MCPHost: return None - async def assemble_memory(self) -> str: - """组装记忆""" + async def assemble_memory(self) -> list[dict[str, str]]: + """组装记忆，返回虚拟的用户与助手间的对话历史""" # 从host的context_list中获取context context_list = self._context_list - return self._env.from_string(MEMORY_TEMPLATE[self._language]).render( - context_list=context_list, - ) + # 构建对话历史列表 + conversation_history = [] + template = self._env.from_string(MEMORY_TEMPLATE[self._language]) + + for index, ctx in enumerate(context_list, start=1): + # 生成user消息（工具调用） + user_message = template.render( + msg_type="user", + step_index=index, + step_description=ctx.step_description, + step_name=ctx.step_name, + input_data=ctx.input_data, + ) + conversation_history.append({ + "role": "user", + "content": user_message.strip(), + }) + + # 生成assistant消息（工具输出） + assistant_message = template.render( + msg_type="assistant", + step_index=index, + step_status=ctx.step_status, + output_data=ctx.output_data, + ) + conversation_history.append({ + "role": "assistant", + "content": assistant_message.strip(), + }) + + return conversation_history async def _save_memory( @@ -102,24 +130,31 @@ class MCPHost: async def _fill_params(self, tool: MCPTools, query: str) -> dict[str, Any]: """填充工具参数""" - llm_query = rf""" - 请使用参数生成工具，生成满足以下目标的工具参数： + # 使用Jinja2模板生成查询 + template = self._env.from_string(FILL_PARAMS_QUERY[self._language]) + llm_query = template.render( + instruction=query, + tool_name=tool.toolName, + tool_description=tool.description, + ) - {query} - """ function_definition = { "name": tool.toolName, "description": tool.description, "parameters": tool.inputSchema, } + # 获取历史对话记录并添加当前查询 + memory_conversation = await self.assemble_memory() + conversation = [ + *memory_conversation, + {"role": "user", "content": llm_query}, + ] + # 使用全局json_generator实例 return await json_generator.generate( function=function_definition, - conversation=[ - {"role": "user", "content": await self.assemble_memory()}, - {"role": "user", "content": llm_query}, - ], + conversation=conversation, language=self._language, ) diff --git a/apps/scheduler/mcp/plan.py b/apps/scheduler/mcp/plan.py index 69b8c2c66..f09967266 100644 --- a/apps/scheduler/mcp/plan.py +++ b/apps/scheduler/mcp/plan.py @@ -1,6 +1,7 @@ # Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. """MCP 用户目标拆解与规划""" +import copy import logging from jinja2 import BaseLoader @@ -10,7 +11,7 @@ from apps.llm import LLM, json_generator from apps.models import LanguageType, MCPTools from apps.schemas.mcp import MCPPlan -from .prompt import CREATE_PLAN, FINAL_ANSWER +from .prompt import CREATE_MCP_PLAN_FUNCTION, CREATE_PLAN, FINAL_ANSWER _logger = logging.getLogger(__name__) @@ -32,15 +33,6 @@ class MCPPlanner: async def create_plan(self, tool_list: list[MCPTools], max_steps: int = 6) -> MCPPlan: """规划下一步的执行流程，并输出""" - # 获取推理结果 - result = await self._get_reasoning_plan(tool_list, max_steps) - - # 解析为结构化数据 - return await self._parse_plan_result(result, max_steps) - - - async def _get_reasoning_plan(self, tool_list: list[MCPTools], max_steps: int) -> str: - """获取推理大模型的结果""" # 格式化Prompt template = self._env.from_string(CREATE_PLAN[self._language]) prompt = template.render( @@ -49,40 +41,18 @@ class MCPPlanner: max_num=max_steps, ) - # 调用推理大模型 - message = [ - {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": prompt}, - ] - result = "" - async for chunk in self._llm.call( - message, - streaming=False, - ): - result += chunk.content or "" - return result - - - async def _parse_plan_result(self, result: str, max_steps: int) -> MCPPlan: - """将推理结果解析为结构化数据""" - # 构造标准 OpenAI Function 格式 - schema = MCPPlan.model_json_schema() - schema["properties"]["plans"]["maxItems"] = max_steps - - function_def = { - "name": "parse_mcp_plan", - "description": ( - "Parse the reasoning result into a structured MCP plan with multiple steps. " - "Each step should include a step ID, content description, tool name, and instruction." - ), - "parameters": schema, - } + # 构造标准 OpenAI Function 格式，使用预定义的 schema 并替换最大条数 + function_def = CREATE_MCP_PLAN_FUNCTION[self._language].copy() + # 深拷贝 parameters 以避免修改原始定义 + function_def["parameters"] = copy.deepcopy(function_def["parameters"]) + # 替换最大条数限制 + function_def["parameters"]["properties"]["plans"]["maxItems"] = max_steps - # 使用全局json_generator实例解析结果 + # 使用json_generator直接生成结构化plan plan = await json_generator.generate( function=function_def, conversation=[ - {"role": "user", "content": result}, + {"role": "user", "content": prompt}, ], language=self._language, ) diff --git a/apps/scheduler/mcp/prompt.py b/apps/scheduler/mcp/prompt.py index bd6246919..85d0fb7e0 100644 --- a/apps/scheduler/mcp/prompt.py +++ b/apps/scheduler/mcp/prompt.py @@ -160,202 +160,101 @@ additional content: CREATE_PLAN: dict[str, str] = { LanguageType.CHINESE: dedent( r""" - 你是一个计划生成器。 - 请分析用户的目标，并生成一个计划。你后续将根据这个计划，一步一步地完成用户的目标。 - - # 一个好的计划应该： - 1. 能够成功完成用户的目标 - 2. 计划中的每一个步骤必须且只能使用一个工具。 - 3. 计划中的步骤必须具有清晰和逻辑的步骤，没有冗余或不必要的步骤。 - 4. 计划中的最后一步必须是Final工具，以确保计划执行结束。 - 5.生成的计划必须要覆盖用户的目标，不能遗漏任何用户目标中的内容。 - - # 生成计划时的注意事项： - - 每一条计划包含3个部分： - - 计划内容：描述单个计划步骤的大致内容 - - 工具ID：必须从下文的工具列表中选择 - - 工具指令：改写用户的目标，使其更符合工具的输入要求 - - 必须按照如下格式生成计划，不要输出任何额外数据： - - ```json - { - "plans": [ - { - "content": "计划内容", - "tool": "工具ID", - "instruction": "工具指令" - } - ] - } - ``` - - - 在生成计划之前，请一步一步思考，解析用户的目标，并指导你接下来的生成。思考过程应按步骤顺序放置在\ - XML标签中。 - - 计划内容中，可以使用"Result[]"来引用之前计划步骤的结果。例如："Result[3]"表示引用第三条计划执行后的结果。 - - 计划不得多于{{ max_num }}条，且每条计划内容应少于150字。 - - # 工具 - 你可以访问并使用一些工具，这些工具将在 XML标签中给出。 - - - {% for tool in tools %} - - {{ tool.id }}{{tool.name}}；{{ tool.description }} - {% endfor %} - - Final结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终\ -结果。 - - - # 样例 - ## 目标 - 在后台运行一个新的alpine:latest容器，将主机/root文件夹挂载至/data，并执行top命令。 - - ## 计划 - - 1. 这个目标需要使用Docker来完成,首先需要选择合适的MCP Server - 2. 目标可以拆解为以下几个部分: - - 运行alpine:latest容器 - - 挂载主机目录 - - 在后台运行 - - 执行top命令 - 3. 需要先选择MCP Server,然后生成Docker命令,最后执行命令 - - - ```json - { - "plans": [ - { - "content": "选择一个支持Docker的MCP Server", - "tool": "mcp_selector", - "instruction": "需要一个支持Docker容器运行的MCP Server" - }, - { - "content": "使用Result[0]中选择的MCP Server，生成Docker命令", - "tool": "command_generator", - "instruction": "生成Docker命令：在后台运行alpine:latest容器，挂载/root到/data，执行top命令" - }, - { - "content": "在Result[0]的MCP Server上执行Result[1]生成的命令", - "tool": "command_executor", - "instruction": "执行Docker命令" - }, - { - "content": "任务执行完成，容器已在后台运行，结果为Result[2]", - "tool": "Final", - "instruction": "" - } - ] - } - ``` - - # 现在开始生成计划： - ## 目标 + ## 任务 + 分析用户目标并生成执行计划，用于后续逐步完成目标。 + + ## 计划要求 + 1. **每步一工具**：每个步骤仅使用一个工具 + 2. **逻辑清晰**：步骤间有明确依赖关系，无冗余 + 3. **目标覆盖**：不遗漏用户目标的任何内容 + 4. **步骤上限**：不超过 {{ max_num }} 条，每条不超过150字 + 5. **必须结束**：最后一步使用 `Final` 工具 + + ## 计划结构 + 每条计划包含三要素： + - **content**：步骤描述，可用 `Result[i]` 引用前序结果 + - **tool**：工具ID（从工具列表中选择） + - **instruction**：针对工具的具体指令 + + ## 工具列表 + {% for tool in tools %} + - **{{ tool.id }}**: {{tool.name}}；{{ tool.description }} + {% endfor %} + - **Final**: 结束步骤，标志计划执行完成 + + ## 示例 + 假设用户目标是：在后台运行 alpine:latest 容器，挂载 /root 到 /data，执行 top 命令 + + 首先分析这个目标： + - 这需要 Docker 支持，因此第一步应该选择一个支持 Docker 的 MCP Server + - 接着需要生成符合要求的 Docker 命令，包括容器镜像、目录挂载、后台运行和命令执行 + - 然后执行生成的命令 + - 最后标记任务完成 + + 基于这个分析，可以生成以下计划： + - 第一步：选择支持 Docker 的 MCP Server，使用 mcp_selector 工具，\ +指令是"需要支持 Docker 容器运行的 MCP Server" + - 第二步：基于 Result[0] 生成 Docker 命令，使用 command_generator 工具，\ +指令是"生成命令：后台运行 alpine:latest，挂载 /root 至 /data，执行 top" + - 第三步：在 Result[0] 上执行 Result[1] 的命令，使用 command_executor 工具，指令是"执行 Docker 命令" + - 第四步：容器已运行，输出 Result[2]，使用 Final 工具，指令为空 + + ## 用户目标 {{goal}} - # 计划 + ## 请生成计划 + 先分析目标，理清思路，然后生成结构化的执行计划。 """, ), LanguageType.ENGLISH: dedent( r""" - You are a plan generator. - Please analyze the user's goal and generate a plan. You will then follow this plan to achieve the user's \ -goal step by step. - - # A good plan should: - 1. Be able to successfully achieve the user's goal. - 2. Each step in the plan must use only one tool. - 3. The steps in the plan must have clear and logical steps, without redundant or unnecessary steps. - 4. The last step in the plan must be a Final tool to ensure that the plan is executed. - - # Things to note when generating plans: - - Each plan contains three parts: - - Plan content: Describes the general content of a single plan step - - Tool ID: Must be selected from the tool list below - - Tool instructions: Rewrite the user's goal to make it more consistent with the tool's input \ -requirements - - Plans must be generated in the following format. Do not output any additional data: - - ```json - { - "plans": [ - { - "content":"Plan content", - "tool":"Tool ID", - "instruction":"Tool instructions" - } - ] - } - ``` - - - Before generating a plan, please think step by step, analyze the user's goal, and guide your next \ -steps. The thought process should be placed in sequential steps within XML tags. - - In the plan content, you can use "Result[]" to reference the results of the previous plan steps. \ -For example: "Result[3]" refers to the result after the third plan is executed. - - The plan should not have more than {{ max_num }} items, and each plan content should be less than \ -150 words. - - # Tools - You can access and use some tools, which will be given in the XML tags. - - - {% for tool in tools %} - - {{ tool.id }}{{tool.name}}; {{ tool.description }} - {% endfor %} - - FinalEnd step. When this step is executed, \ - Indicates that the plan execution is completed and the result obtained will be used as the final \ -result. - - - # Example - ## Target - Run a new alpine:latest container in the background, mount the host/root folder to /data, and execute the \ -top command. - - ## Plan - - 1. This goal needs to be completed using Docker. First, you need to select a suitable MCP Server. - 2. The goal can be broken down into the following parts: - - Run the alpine:latest container - - Mount the host directory - - Run in the background - - Execute the top command - 3. You need to select an MCP Server first, then generate the Docker command, and finally execute the \ -command. - - - ```json - { - "plans": [ - { - "content": "Select an MCP Server that supports Docker", - "tool": "mcp_selector", - "instruction": "You need an MCP Server that supports running Docker containers" - }, - { - "content": "Use the MCP Server selected in Result[0] to generate Docker commands", - "tool": "command_generator", - "instruction": "Generate Docker command: Run the alpine:latest container in the background, \ -mount /root to /data, and execute the top command" - }, - { - "content": "Execute the command generated by Result[1] on the MCP Server of Result[0]", - "tool": "command_executor", - "instruction": "Execute Docker command" - }, - { - "content": "Task execution completed, the container is running in the background, the result \ -is Result[2]", - "tool": "Final", - "instruction": "" - } - ] - } - ``` + ## Task + Analyze the user's goal and generate an execution plan to accomplish the goal step by step. + + ## Plan Requirements + 1. **One Tool Per Step**: Each step uses only one tool + 2. **Clear Logic**: Steps have explicit dependencies, no redundancy + 3. **Complete Coverage**: Don't miss any part of the user's goal + 4. **Step Limit**: Maximum {{ max_num }} steps, each under 150 words + 5. **Must End**: Final step must use the `Final` tool + + ## Plan Structure + Each plan contains three elements: + - **content**: Step description, can reference previous results with `Result[i]` + - **tool**: Tool ID (selected from tool list) + - **instruction**: Specific instruction for the tool + + ## Tool List + {% for tool in tools %} + - **{{ tool.id }}**: {{tool.name}}; {{ tool.description }} + {% endfor %} + - **Final**: End step, marks plan execution complete - # Now start generating the plan: - ## Goal + ## Example + Suppose the user's goal is: Run alpine:latest container in background, mount /root to /data, \ +execute top command + + First, analyze this goal: + - This requires Docker support, so the first step should select an MCP Server with Docker capability + - Next, need to generate a Docker command that meets the requirements, including container image, \ +directory mount, background execution, and command execution + - Then execute the generated command + - Finally, mark the task as complete + + Based on this analysis, the following plan can be generated: + - Step 1: Select MCP Server with Docker support, using mcp_selector tool, \ +instruction is "Need MCP Server supporting Docker container operation" + - Step 2: Generate Docker command based on Result[0], using command_generator tool, \ +instruction is "Generate command: run alpine:latest in background, mount /root to /data, execute top" + - Step 3: Execute Result[1] command on Result[0], using command_executor tool, \ +instruction is "Execute Docker command" + - Step 4: Container running, output Result[2], using Final tool, instruction is empty + + ## User Goal {{goal}} - # Plan + ## Please Generate Plan + First analyze the goal and clarify your thinking, then generate a structured execution plan. """, ), } @@ -494,22 +393,26 @@ the completion status of the goal. MEMORY_TEMPLATE: dict[str, str] = { LanguageType.CHINESE: dedent( r""" - {% for ctx in context_list %} - - 第{{ loop.index }}步：{{ ctx.step_description }} - - 调用工具 `{{ ctx.step_name }}`，并提供参数 `{{ ctx.input_data | tojson }}`。 - - 执行状态：{{ ctx.step_status }} - - 得到数据：`{{ ctx.output_data | tojson }}` - {% endfor %} + {% if msg_type == "user" %} + 第{{ step_index }}步：{{ step_description }} + 调用工具 `{{ step_name }}`，并提供参数 `{{ input_data | tojson }}`。 + {% elif msg_type == "assistant" %} + 第{{ step_index }}步执行完成。 + 执行状态：{{ step_status }} + 得到数据：`{{ output_data | tojson }}` + {% endif %} """, ), LanguageType.ENGLISH: dedent( r""" - {% for ctx in context_list %} - - Step {{ loop.index }}: {{ ctx.step_description }} - - Called tool `{{ ctx.step_name }}` and provided parameters `{{ ctx.input_data }}` - - Execution status: {{ ctx.step_status }} - - Got data: `{{ ctx.output_data }}` - {% endfor %} + {% if msg_type == "user" %} + Step {{ step_index }}: {{ step_description }} + Called tool `{{ step_name }}` and provided parameters `{{ input_data | tojson }}` + {% elif msg_type == "assistant" %} + Step {{ step_index }} execution completed. + Execution status: {{ step_status }} + Got data: `{{ output_data | tojson }}` + {% endif %} """, ), } @@ -546,3 +449,204 @@ MCP_FUNCTION_SELECT: dict[LanguageType, str] = { """, ), } + +FILL_PARAMS_QUERY: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" + ## 当前目标 + + {{ instruction }} + + ## 可用工具 + + 你可以使用 `{{ tool_name }}` 工具来完成上述目标。 + + ### 工具信息 + - **工具名称**: `{{ tool_name }}` + - **工具描述**: {{ tool_description }} + + ## 说明 + + 请调用上述工具来完成当前目标。注意事项： + + 1. 仔细分析目标的具体要求，确保工具调用能达成预期结果 + 2. 参考上下文信息，合理使用已有的结果和数据 + 3. 提供所有必需信息，可选信息根据需要补充 + 4. 必须至少选择1个工具进行调用 + 5. 不要编造或假设不存在的信息 + """, + ), + LanguageType.ENGLISH: dedent( + r""" + ## Current Objective + + {{ instruction }} + + ## Available Tool + + You can use the `{{ tool_name }}` tool to accomplish the above objective. + + ### Tool Information + - **Tool Name**: `{{ tool_name }}` + - **Tool Description**: {{ tool_description }} + + ## Instructions + + Please invoke the above tool to complete the current objective. Key points: + + 1. Carefully analyze the objective requirements to ensure the tool invocation achieves the expected results + 2. Reference contextual information and reasonably use existing results and data + 3. Provide all required information, supplement optional information as needed + 4. Must invoke at least 1 tool + 5. Do not fabricate or assume non-existent information + """, + ), +} + +CREATE_MCP_PLAN_FUNCTION: dict[LanguageType, dict] = { + LanguageType.CHINESE: { + "name": "create_mcp_plan", + "description": "生成结构化的MCP计划以实现用户目标。每个步骤应包含步骤ID、内容描述、工具名称和指令。", + "parameters": { + "type": "object", + "properties": { + "plans": { + "type": "array", + "description": ( + "按顺序执行的计划步骤列表。每个步骤使用一个工具，" + "并描述要做什么、使用哪个工具以及具体指令。" + ), + "items": { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": "该步骤的描述。可以使用 Result[i] 语法引用之前的结果。应少于150字。", + }, + "tool": { + "type": "string", + "description": ( + "该步骤使用的工具ID。必须从提供的工具列表中选择。" + "最后一步必须使用 'Final' 工具。" + ), + }, + "instruction": { + "type": "string", + "description": "工具的具体指令。描述该工具在此步骤中应执行的操作。", + }, + }, + "required": ["content", "tool", "instruction"], + }, + "minItems": 1, + "maxItems": 10, + }, + }, + "required": ["plans"], + "additionalProperties": False, + }, + "examples": [ + { + "plans": [ + { + "content": "选择支持 Docker 的 MCP Server", + "tool": "mcp_selector", + "instruction": "需要支持 Docker 容器运行的 MCP Server", + }, + { + "content": "基于 Result[0] 生成 Docker 命令", + "tool": "command_generator", + "instruction": "生成命令：后台运行 alpine:latest，挂载 /root 至 /data，执行 top", + }, + { + "content": "在 Result[0] 上执行 Result[1] 的命令", + "tool": "command_executor", + "instruction": "执行 Docker 命令", + }, + { + "content": "容器已运行，输出 Result[2]", + "tool": "Final", + "instruction": "", + }, + ], + }, + ], + }, + LanguageType.ENGLISH: { + "name": "create_mcp_plan", + "description": ( + "Generate a structured MCP plan to achieve the user's goal. " + "Each step should include a step ID, content description, tool name, and instruction." + ), + "parameters": { + "type": "object", + "properties": { + "plans": { + "type": "array", + "description": ( + "List of plan steps to execute sequentially. Each step uses one tool and " + "describes what to do, which tool to use, and specific instructions." + ), + "items": { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": ( + "Description of this step. Can reference previous results using " + "Result[i] syntax. Should be under 150 words." + ), + }, + "tool": { + "type": "string", + "description": ( + "Tool ID to use for this step. Must be selected from the provided " + "tool list. The final step must use 'Final' tool." + ), + }, + "instruction": { + "type": "string", + "description": ( + "Specific instruction for the tool. Describes what the tool should do in this step." + ), + }, + }, + "required": ["content", "tool", "instruction"], + }, + "minItems": 1, + "maxItems": 10, + }, + }, + "required": ["plans"], + "additionalProperties": False, + }, + "examples": [ + { + "plans": [ + { + "content": "Select MCP Server with Docker support", + "tool": "mcp_selector", + "instruction": "Need MCP Server supporting Docker container operation", + }, + { + "content": "Generate Docker command based on Result[0]", + "tool": "command_generator", + "instruction": ( + "Generate command: run alpine:latest in background, " + "mount /root to /data, execute top" + ), + }, + { + "content": "Execute Result[1] command on Result[0]", + "tool": "command_executor", + "instruction": "Execute Docker command", + }, + { + "content": "Container running, output Result[2]", + "tool": "Final", + "instruction": "", + }, + ], + }, + ], + }, +} -- Gitee