diff --git a/apps/scheduler/call/facts/facts.py b/apps/scheduler/call/facts/facts.py
index 5f2bffbec3d01d723b86a98b5a1128278e9941de..5401531abed4f0e4c0c39bf7eadbe24e78a6dfbc 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 e4cfb9b7ff3bb466d23a8e1fbbe38734230c317d..22fab49f8e68479e699c2463134fe937645ade11 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 b4d849d2fc8ad08e11f96467a0045f86c9cf0dc3..cd6560fe6ca62726f1c4c27ed52b4122ddb0f2db 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 59c0eadbe7dad77bedf1dc7546201a7bbcc05412..4058a3e2b26b54676ef52ebd3016b07641cdbff4 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 e90985e0d5a600ebab4983982319f07f078a4e51..3cbf9b0a58a6633a4a6ee910100c0429e57b057d 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 16505d5001688ef35b2cda1e5b5a3c15ac473431..143569f641230a153f30dbec849c4ad6640dddd2 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 33285ebc4c0ef1229ba6078b7ad2af863bcc3bb1..8ae8b5051d841363ad605960de920c6850f4e317 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 69b8c2c66f61e1349b65a0caac6d3d3d84f48d3d..f0996726688378700d14b8aea628262484c0ae66 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 bd6246919b69904d5872f0a73e42a989e7d2ad6c..85d0fb7e033d2d1ca16786ca0bad9ea01619e06e 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": "",
+ },
+ ],
+ },
+ ],
+ },
+}