diff --git a/apps/scheduler/call/api/api.py b/apps/scheduler/call/api/api.py index e1891f7259b72a2fa03228f6289c54da6297b958..66aa838690f25fc9dbc52e35cb1850f2ba4189ba 100644 --- a/apps/scheduler/call/api/api.py +++ b/apps/scheduler/call/api/api.py @@ -5,7 +5,7 @@ import json import logging from collections.abc import AsyncGenerator from functools import partial -from typing import Any +from typing import Any, ClassVar import httpx from fastapi import status @@ -15,7 +15,7 @@ from pydantic.json_schema import SkipJsonSchema from apps.common.oidc import oidc_provider from apps.scheduler.call.api.schema import APIInput, APIOutput from apps.scheduler.call.core import CoreCall -from apps.schemas.enum_var import CallOutputType, ContentType, HTTPMethod +from apps.schemas.enum_var import CallOutputType, ContentType, HTTPMethod, LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -59,10 +59,27 @@ class API(CoreCall, input_model=APIInput, output_model=APIOutput): body: dict[str, Any] = Field(description="已知的部分请求体", default={}) query: dict[str, Any] = Field(description="已知的部分请求参数", default={}) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "API调用", + "description": "向某一个API接口发送HTTP请求，获取数据", + }, + LanguageType.ENGLISH: { + "name": "API Call", + "description": "Send an HTTP request to an API to obtain data", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="API调用", description="向某一个API接口发送HTTP请求，获取数据。") + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> APIInput: """初始化API调用工具""" @@ -99,8 +116,10 @@ class API(CoreCall, input_model=APIInput, output_model=APIOutput): body=self.body, ) - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: - """调用API，然后返回LLM解析后的数据""" + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: + """调用API，然后返回LLM解析后的数据""" self._client = httpx.AsyncClient(timeout=self.timeout) input_obj = APIInput.model_validate(input_data) try: @@ -112,7 +131,9 @@ class API(CoreCall, input_model=APIInput, output_model=APIOutput): finally: await self._client.aclose() - async def _make_api_call(self, data: APIInput, files: dict[str, tuple[str, bytes, str]]) -> httpx.Response: + async def _make_api_call( + self, data: APIInput, files: dict[str, tuple[str, bytes, str]] + ) -> httpx.Response: """组装API请求""" # 获取必要参数 if self._auth: diff --git a/apps/scheduler/call/choice/choice.py b/apps/scheduler/call/choice/choice.py index 01ac7106fbdae673d12488a49d831f1d7e7fc13d..38e9c0af4b6a6b9ae85983c6abb689818b8de6a6 100644 --- a/apps/scheduler/call/choice/choice.py +++ b/apps/scheduler/call/choice/choice.py @@ -5,7 +5,7 @@ import ast import copy import logging from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, ClassVar from pydantic import Field @@ -15,11 +15,12 @@ from apps.scheduler.call.choice.schema import ( ChoiceBranch, ChoiceInput, ChoiceOutput, + Condition, Logic, ) -from apps.schemas.parameters import Type from apps.scheduler.call.core import CoreCall -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType +from apps.schemas.parameters import Type from apps.schemas.scheduler import ( CallError, CallInfo, @@ -34,13 +35,36 @@ class Choice(CoreCall, input_model=ChoiceInput, output_model=ChoiceOutput): """Choice工具""" to_user: bool = Field(default=False) - choices: list[ChoiceBranch] = Field(description="分支", default=[ChoiceBranch(), - ChoiceBranch(conditions=[Condition()], is_default=False)]) + choices: list[ChoiceBranch] = Field( + description="分支", default=[ChoiceBranch(), ChoiceBranch(conditions=[Condition()], is_default=False)] + ) + + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "判断", + "description": "使用大模型或使用程序做出判断", + }, + LanguageType.ENGLISH: { + "name": "Choice", + "description": "Use a large model or a program to make a decision", + }, + } @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="选择器", description="使用大模型或使用程序做出判断") + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) + + def _raise_value_error(self, msg: str) -> None: + """统一处理 ValueError 异常抛出""" + logger.warning(msg) + raise ValueError(msg) async def _prepare_message(self, call_vars: CallVars) -> list[dict[str, Any]]: """替换choices中的系统变量""" @@ -135,7 +159,7 @@ class Choice(CoreCall, input_model=ChoiceInput, output_model=ChoiceOutput): ) async def _exec( - self, input_data: dict[str, Any] + self, input_data: dict[str, Any], language: LanguageType ) -> AsyncGenerator[CallOutputChunk, None]: """执行Choice工具""" # 解析输入数据 diff --git a/apps/scheduler/call/choice/schema.py b/apps/scheduler/call/choice/schema.py index d97a0c8d7e9efcef255ced76c8d1d6b39dfed241..955322705031604531cc14f6fe7a4a30935d5989 100644 --- a/apps/scheduler/call/choice/schema.py +++ b/apps/scheduler/call/choice/schema.py @@ -1,20 +1,19 @@ # Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. """Choice Call的输入和输出""" import uuid - from enum import Enum -from pydantic import BaseModel, Field +from pydantic import Field +from apps.scheduler.call.core import DataBase from apps.schemas.parameters import ( - Type, - NumberOperate, - StringOperate, - ListOperate, BoolOperate, DictOperate, + ListOperate, + NumberOperate, + StringOperate, + Type, ) -from apps.scheduler.call.core import DataBase class Logic(str, Enum): diff --git a/apps/scheduler/call/core.py b/apps/scheduler/call/core.py index 5bed80309fbdb993aacc378ab0319ed291f1ffb3..16db721550a41c73d2dc2b5fda14ce29821b5438 100644 --- a/apps/scheduler/call/core.py +++ b/apps/scheduler/call/core.py @@ -14,7 +14,7 @@ from pydantic.json_schema import SkipJsonSchema from apps.llm.function import FunctionLLM from apps.llm.reasoning import ReasoningLLM -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.pool import NodePool from apps.schemas.scheduler import ( CallError, @@ -25,6 +25,7 @@ from apps.schemas.scheduler import ( CallVars, ) from apps.schemas.task import FlowStepHistory +from apps.schemas.enum_var import LanguageType if TYPE_CHECKING: from apps.scheduler.executor.step import StepExecutor @@ -52,7 +53,9 @@ class CoreCall(BaseModel): name: SkipJsonSchema[str] = Field(description="Step的名称", exclude=True) description: SkipJsonSchema[str] = Field(description="Step的描述", exclude=True) node: SkipJsonSchema[NodePool | None] = Field(description="节点信息", exclude=True) - enable_filling: SkipJsonSchema[bool] = Field(description="是否需要进行自动参数填充", default=False, exclude=True) + enable_filling: SkipJsonSchema[bool] = Field( + description="是否需要进行自动参数填充", default=False, exclude=True + ) tokens: SkipJsonSchema[CallTokens] = Field( description="Call的输入输出Tokens信息", default=CallTokens(), @@ -68,6 +71,12 @@ class CoreCall(BaseModel): exclude=True, frozen=True, ) + language: ClassVar[SkipJsonSchema[LanguageType]] = Field( + description="语言", + default=LanguageType.CHINESE, + exclude=True, + ) + i18n_info: ClassVar[SkipJsonSchema[dict[str, dict]]] = {} to_user: bool = Field(description="是否需要将输出返回给用户", default=False) @@ -76,7 +85,9 @@ class CoreCall(BaseModel): extra="allow", ) - def __init_subclass__(cls, input_model: type[DataBase], output_model: type[DataBase], **kwargs: Any) -> None: + def __init_subclass__( + cls, input_model: type[DataBase], output_model: type[DataBase], **kwargs: Any + ) -> None: """初始化子类""" super().__init_subclass__(**kwargs) cls.input_model = input_model @@ -181,9 +192,14 @@ class CoreCall(BaseModel): async def _after_exec(self, input_data: dict[str, Any]) -> None: """Call类实例的执行后方法""" - async def exec(self, executor: "StepExecutor", input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def exec( + self, + executor: "StepExecutor", + input_data: dict[str, Any], + language: LanguageType = LanguageType.CHINESE, + ) -> AsyncGenerator[CallOutputChunk, None]: """Call类实例的执行方法""" - async for chunk in self._exec(input_data): + async for chunk in self._exec(input_data, language): yield chunk await self._after_exec(input_data) diff --git a/apps/scheduler/call/empty.py b/apps/scheduler/call/empty.py index 5865bc7e804491a7d6aa41fa251dc8c5d9c77dc6..590a24b9372cc492670323161c9e0d7e87c83e11 100644 --- a/apps/scheduler/call/empty.py +++ b/apps/scheduler/call/empty.py @@ -5,7 +5,7 @@ from collections.abc import AsyncGenerator from typing import Any from apps.scheduler.call.core import CoreCall, DataBase -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.scheduler import CallInfo, CallOutputChunk, CallVars @@ -34,7 +34,7 @@ class Empty(CoreCall, input_model=DataBase, output_model=DataBase): return DataBase() - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec(self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE) -> AsyncGenerator[CallOutputChunk, None]: """ 执行Call diff --git a/apps/scheduler/call/facts/facts.py b/apps/scheduler/call/facts/facts.py index 2b9df0c6f8f1002e76bda24eb099f47a6084b2af..42c3c7f1e8cf8077d1d5e58397ef0131af909c7a 100644 --- a/apps/scheduler/call/facts/facts.py +++ b/apps/scheduler/call/facts/facts.py @@ -2,7 +2,7 @@ """提取事实工具""" from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, ClassVar from jinja2 import BaseLoader from jinja2.sandbox import SandboxedEnvironment @@ -16,7 +16,7 @@ from apps.scheduler.call.facts.schema import ( FactsInput, FactsOutput, ) -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.pool import NodePool from apps.schemas.scheduler import CallInfo, CallOutputChunk, CallVars from apps.services.user_domain import UserDomainManager @@ -30,10 +30,27 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): answer: str = Field(description="用户输入") + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "提取事实", + "description": "从对话上下文和文档片段中提取事实。", + }, + LanguageType.ENGLISH: { + "name": "Fact Extraction", + "description": "Extract facts from the conversation context and document snippets.", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="提取事实", description="从对话上下文和文档片段中提取事实。") + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) @classmethod async def instance(cls, executor: "StepExecutor", node: NodePool | None, **kwargs: Any) -> Self: @@ -62,7 +79,9 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): message=message, ) - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """执行工具""" data = FactsInput(**input_data) # jinja2 环境 @@ -74,7 +93,7 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): ) # 提取事实信息 - facts_tpl = env.from_string(FACTS_PROMPT) + facts_tpl = env.from_string(FACTS_PROMPT[language]) facts_prompt = facts_tpl.render(conversation=data.message) facts_obj: FactsGen = await self._json([ {"role": "system", "content": "You are a helpful assistant."}, @@ -82,7 +101,7 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): ], FactsGen) # type: ignore[arg-type] # 更新用户画像 - domain_tpl = env.from_string(DOMAIN_PROMPT) + domain_tpl = env.from_string(DOMAIN_PROMPT[language]) domain_prompt = domain_tpl.render(conversation=data.message) domain_list: DomainGen = await self._json([ {"role": "system", "content": "You are a helpful assistant."}, @@ -100,7 +119,12 @@ class FactsCall(CoreCall, input_model=FactsInput, output_model=FactsOutput): ).model_dump(by_alias=True, exclude_none=True), ) - async def exec(self, executor: "StepExecutor", input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def exec( + self, + executor: "StepExecutor", + input_data: dict[str, Any], + language: LanguageType = LanguageType.CHINESE, + ) -> AsyncGenerator[CallOutputChunk, None]: """执行工具""" async for chunk in self._exec(input_data): content = chunk.content diff --git a/apps/scheduler/call/facts/prompt.py b/apps/scheduler/call/facts/prompt.py index b2b2513f2c28feb4d19f5fb4ee3eba1bd616a4dc..02e134391fa33d3dbe3b3e71cc63947acc739777 100644 --- a/apps/scheduler/call/facts/prompt.py +++ b/apps/scheduler/call/facts/prompt.py @@ -2,8 +2,10 @@ """记忆提取工具的提示词""" from textwrap import dedent - -DOMAIN_PROMPT: str = dedent(r""" +from apps.schemas.enum_var import LanguageType +DOMAIN_PROMPT: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 根据对话上文，提取推荐系统所需的关键词标签，要求： @@ -35,8 +37,48 @@ DOMAIN_PROMPT: str = dedent(r""" {% endfor %} -""") -FACTS_PROMPT: str = dedent(r""" +""" + ), + LanguageType.ENGLISH: dedent( + r""" + + + Extract keywords for recommendation system based on the previous conversation, requirements: + 1. Entity nouns, technical terms, time range, location, product, etc. can be keyword tags + 2. At least one keyword is related to the topic of the conversation + 3. Tags should be concise and not repeated, not exceeding 10 characters + 4. Output in JSON format, do not include XML tags, do not include any explanatory notes + + + + + What's the weather like in Beijing? + Beijing is sunny today. + + + + { + "keywords": ["Beijing", "weather"] + } + + + + + + {% for item in conversation %} + <{{item['role']}}> + {{item['content']}} + + {% endfor %} + + +""" + ), +} + +FACTS_PROMPT: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 从对话中提取关键信息，并将它们组织成独一无二的、易于理解的事实，包含用户偏好、关系、实体等有用信息。 @@ -80,4 +122,53 @@ FACTS_PROMPT: str = dedent(r""" {% endfor %} -""") +""" + ), + LanguageType.ENGLISH: dedent( + r""" + + + Extract key information from the conversation and organize it into unique, easily understandable facts, including user preferences, relationships, entities, etc. + The following are the types of information you need to pay attention to and detailed instructions on how to handle input data. + + **Types of information you need to pay attention to** + 1. Entities: Entities involved in the conversation. For example: names, locations, organizations, events, etc. + 2. Preferences: Attitudes towards entities. For example: like, dislike, etc. + 3. Relationships: Relationships between users and entities, or between two entities. For example: include, parallel, mutually exclusive, etc. + 4. Actions: Specific actions that affect entities. For example: query, search, browse, click, etc. + + **Requirements** + 1. Facts must be accurate and can only be extracted from the conversation. Do not include the information in the example in the output. + 2. Facts must be clear, concise, and easy to understand. Must be less than 30 words. + 3. Output in the following JSON format: + + { + "facts": ["Fact 1", "Fact 2", "Fact 3"] + } + + + + + What are the attractions in Hangzhou West Lake? + West Lake in Hangzhou, Zhejiang Province, China, is a famous scenic spot known for its beautiful natural scenery and rich cultural heritage. Many notable attractions surround West Lake, including the renowned Su Causeway, Bai Causeway, Broken Bridge, and the Three Pools Mirroring the Moon. Famous for its crystal-clear waters and the surrounding mountains, West Lake is one of China's most famous lakes. + + + + { + "facts": ["Hangzhou West Lake has famous attractions such as Suzhou Embankment, Bai Budi, Qiantang Bridge, San Tang Yue, etc."] + } + + + + + + {% for item in conversation %} + <{{item['role']}}> + {{item['content']}} + + {% endfor %} + + +""" + ), +} diff --git a/apps/scheduler/call/graph/graph.py b/apps/scheduler/call/graph/graph.py index c2728f17913fcd0e8343168f2b508dbe6006fd6e..c4cd6b68515904696d113fa1ea896dba362d3799 100644 --- a/apps/scheduler/call/graph/graph.py +++ b/apps/scheduler/call/graph/graph.py @@ -3,7 +3,7 @@ import json from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, ClassVar from anyio import Path from pydantic import Field @@ -11,7 +11,7 @@ from pydantic import Field from apps.scheduler.call.core import CoreCall from apps.scheduler.call.graph.schema import RenderFormat, RenderInput, RenderOutput from apps.scheduler.call.graph.style import RenderStyle -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -25,12 +25,27 @@ class Graph(CoreCall, input_model=RenderInput, output_model=RenderOutput): dataset_key: str = Field(description="图表的数据来源（字段名）", default="") + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "图表", + "description": "将SQL查询出的数据转换为图表。", + }, + LanguageType.ENGLISH: { + "name": "Chart", + "description": "Convert the data queried by SQL into a chart.", + }, + } @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="图表", description="将SQL查询出的数据转换为图表") + """ + 返回Call的名称和描述 + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> RenderInput: """初始化Render Call，校验参数，读取option模板""" @@ -54,8 +69,9 @@ class Graph(CoreCall, input_model=RenderInput, output_model=RenderOutput): data=data, ) - - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """运行Render Call""" data = RenderInput(**input_data) @@ -117,8 +133,9 @@ class Graph(CoreCall, input_model=RenderInput, output_model=RenderOutput): result.append({"type": key, "value": val}) return result - - def _parse_options(self, column_num: int, chart_style: str, additional_style: str, scale_style: str) -> None: + def _parse_options( + self, column_num: int, chart_style: str, additional_style: str, scale_style: str + ) -> None: """解析LLM做出的图表样式选择""" series_template = {} diff --git a/apps/scheduler/call/llm/llm.py b/apps/scheduler/call/llm/llm.py index 6a679dce98af6164211edd16b2fa38714d899f31..e1766674e4ac48fdf7f1e121627fceda1bd66bcb 100644 --- a/apps/scheduler/call/llm/llm.py +++ b/apps/scheduler/call/llm/llm.py @@ -4,7 +4,7 @@ import logging from collections.abc import AsyncGenerator from datetime import datetime -from typing import Any +from typing import Any, ClassVar import pytz from jinja2 import BaseLoader @@ -15,7 +15,7 @@ from apps.llm.reasoning import ReasoningLLM from apps.scheduler.call.core import CoreCall from apps.scheduler.call.llm.prompt import LLM_CONTEXT_PROMPT, LLM_DEFAULT_PROMPT from apps.scheduler.call.llm.schema import LLMInput, LLMOutput -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -38,12 +38,27 @@ class LLM(CoreCall, input_model=LLMInput, output_model=LLMOutput): system_prompt: str = Field(description="大模型系统提示词", default="You are a helpful assistant.") user_prompt: str = Field(description="大模型用户提示词", default=LLM_DEFAULT_PROMPT) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "大模型", + "description": "以指定的提示词和上下文信息调用大模型，并获得输出。", + }, + LanguageType.ENGLISH: { + "name": "Foundation Model", + "description": "Call the foundation model with specified prompt and context, and obtain the output.", + }, + } @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="大模型", description="以指定的提示词和上下文信息调用大模型，并获得输出。") + """ + 返回Call的名称和描述 + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _prepare_message(self, call_vars: CallVars) -> list[dict[str, Any]]: """准备消息""" @@ -100,8 +115,9 @@ class LLM(CoreCall, input_model=LLMInput, output_model=LLMOutput): message=await self._prepare_message(call_vars), ) - - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """运行LLM Call""" data = LLMInput(**input_data) try: diff --git a/apps/scheduler/call/llm/prompt.py b/apps/scheduler/call/llm/prompt.py index 0f227dcaa618b11a2c888f55a61fa51f349d7d8b..20536ff27b28b17979bbf7eed3b94f5e15f2139a 100644 --- a/apps/scheduler/call/llm/prompt.py +++ b/apps/scheduler/call/llm/prompt.py @@ -2,16 +2,34 @@ """大模型工具的提示词""" from textwrap import dedent +from apps.schemas.enum_var import LanguageType LLM_CONTEXT_PROMPT = dedent( + # r""" + # 以下是对用户和AI间对话的简短总结，在中给出： + # + # {{ summary }} + # + # 你作为AI，在回答用户的问题前，需要获取必要的信息。为此，你调用了一些工具，并获得了它们的输出： + # 工具的输出数据将在中给出， 其中为工具的名称，为工具的输出数据。 + # + # {% for tool in history_data %} + # + # {{ tool.step_name }} + # {{ tool.step_description }} + # {{ tool.output_data }} + # + # {% endfor %} + # + # """, r""" - 以下是对用户和AI间对话的简短总结，在中给出： + The following is a brief summary of the user and AI conversation, given in : {{ summary }} - 你作为AI，在回答用户的问题前，需要获取必要的信息。为此，你调用了一些工具，并获得了它们的输出： - 工具的输出数据将在中给出， 其中为工具的名称，为工具的输出数据。 + As an AI, before answering the user's question, you need to obtain necessary information. For this purpose, you have called some tools and obtained their outputs: + The output data of the tools will be given in , where is the name of the tool and is the output data of the tool. {% for tool in history_data %} @@ -24,12 +42,29 @@ LLM_CONTEXT_PROMPT = dedent( """, ).strip("\n") LLM_DEFAULT_PROMPT = dedent( + # r""" + # + # 你是一个乐于助人的智能助手。请结合给出的背景信息, 回答用户的提问。 + # 当前时间：{{ time }}，可以作为时间参照。 + # 用户的问题将在中给出，上下文背景信息将在中给出。 + # 注意：输出不要包含任何XML标签，不要编造任何信息。若你认为用户提问与背景信息无关，请忽略背景信息直接作答。 + # + # + # {{ question }} + # + # + # {{ context }} + # + # 现在，输出你的回答： + # """, r""" - 你是一个乐于助人的智能助手。请结合给出的背景信息, 回答用户的提问。 - 当前时间：{{ time }}，可以作为时间参照。 - 用户的问题将在中给出，上下文背景信息将在中给出。 - 注意：输出不要包含任何XML标签，不要编造任何信息。若你认为用户提问与背景信息无关，请忽略背景信息直接作答。 + You are a helpful AI assistant. Please answer the user's question based on the given background information. + Current time: {{ time }}, which can be used as a reference. + The user's question will be given in , and the context background information will be given in . + + Respond using the same language as the user's question, unless the user explicitly requests a specific language—then follow that request. + Note: Do not include any XML tags in the output. Do not make up any information. If you think the user's question is unrelated to the background information, please ignore the background information and answer directly. @@ -39,12 +74,13 @@ LLM_DEFAULT_PROMPT = dedent( {{ context }} - - 现在，输出你的回答： - """, + Now, please output your answer: + """ ).strip("\n") -LLM_ERROR_PROMPT = dedent( - r""" + +LLM_ERROR_PROMPT = { + LanguageType.CHINESE: dedent( + r""" 你是一位智能助手，能够根据用户的问题，使用Python工具获取信息，并作出回答。你在使用工具解决回答用户的问题时，发生了错误。 你的任务是：分析工具（Python程序）的异常信息，分析造成该异常可能的原因，并以通俗易懂的方式，将原因告知用户。 @@ -67,8 +103,36 @@ LLM_ERROR_PROMPT = dedent( 现在，输出你的回答： - """, -).strip("\n") + """ + ).strip("\n"), + LanguageType.ENGLISH: dedent( + r""" + + You are an intelligent assistant. When using Python tools to answer user questions, an error occurred. + Your task is: Analyze the exception information of the tool (Python program), analyze the possible causes of the error, and inform the user in an easy-to-understand way. + + Current time: {{ time }}, which can be used as a reference. + The program exception information that occurred will be given in , the user's question will be given in , and the context background information will be given in . + Note: Do not include any XML tags in the output. Do not make up any information. If you think the user's question is unrelated to the background information, please ignore the background information. + + + + {{ error_info }} + + + + {{ question }} + + + + {{ context }} + + + Now, please output your answer: + """ + ).strip("\n"), +} + RAG_ANSWER_PROMPT = dedent( r""" diff --git a/apps/scheduler/call/mcp/mcp.py b/apps/scheduler/call/mcp/mcp.py index bd0257b49fb6ea24fbbc429cb11f7b8e6be364b9..07d23aaf60397ee2758e2412fc5dfe4b5daf60ed 100644 --- a/apps/scheduler/call/mcp/mcp.py +++ b/apps/scheduler/call/mcp/mcp.py @@ -4,7 +4,7 @@ import logging from collections.abc import AsyncGenerator from copy import deepcopy -from typing import Any +from typing import Any, ClassVar from pydantic import Field @@ -16,7 +16,7 @@ from apps.scheduler.call.mcp.schema import ( MCPOutput, ) from apps.scheduler.mcp import MCPHost, MCPPlanner, MCPSelector -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.mcp import MCPPlanItem from apps.schemas.scheduler import ( CallInfo, @@ -26,6 +26,28 @@ from apps.schemas.scheduler import ( logger = logging.getLogger(__name__) +MCP_GENERATE: dict[str, dict[LanguageType, str]] = { + "START": { + LanguageType.CHINESE: "[MCP] 开始生成计划...\n\n\n\n", + LanguageType.ENGLISH: "[MCP] Start generating plan...\n\n\n\n", + }, + "END": { + LanguageType.CHINESE: "[MCP] 计划生成完成：\n\n{plan_str}\n\n\n\n", + LanguageType.ENGLISH: "[MCP] Plan generation completed: \n\n{plan_str}\n\n\n\n", + }, +} + +MCP_SUMMARY: dict[str, dict[LanguageType, str]] = { + "START": { + LanguageType.CHINESE: "[MCP] 正在总结任务结果...\n\n", + LanguageType.ENGLISH: "[MCP] Start summarizing task results...\n\n", + }, + "END": { + LanguageType.CHINESE: "[MCP] 任务完成\n\n---\n\n{answer}\n\n", + LanguageType.ENGLISH: "[MCP] Task summary completed\n\n{answer}\n\n", + }, +} + class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): """MCP工具""" @@ -35,6 +57,17 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): text_output: bool = Field(description="是否将结果以文本形式返回", default=True) to_user: bool = Field(description="是否将结果返回给用户", default=True) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "MCP", + "description": "调用MCP Server，执行工具", + }, + LanguageType.ENGLISH: { + "name": "MCP", + "description": "Call the MCP Server to execute tools", + }, + } + @classmethod def info(cls) -> CallInfo: """ @@ -43,7 +76,8 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): :return: Call的名称和描述 :rtype: CallInfo """ - return CallInfo(name="MCP", description="调用MCP Server，执行工具") + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> MCPInput: """初始化MCP""" @@ -61,31 +95,33 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): return MCPInput(avaliable_tools=avaliable_tools, max_steps=self.max_steps) - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """执行MCP""" # 生成计划 - async for chunk in self._generate_plan(): + async for chunk in self._generate_plan(language): yield chunk # 执行计划 plan_list = deepcopy(self._plan.plans) while len(plan_list) > 0: - async for chunk in self._execute_plan_item(plan_list.pop(0)): + async for chunk in self._execute_plan_item(plan_list.pop(0), language): yield chunk # 生成总结 - async for chunk in self._generate_answer(): + async for chunk in self._generate_answer(language): yield chunk - async def _generate_plan(self) -> AsyncGenerator[CallOutputChunk, None]: + async def _generate_plan(self, language) -> AsyncGenerator[CallOutputChunk, None]: """生成执行计划""" # 开始提示 - yield self._create_output("[MCP] 开始生成计划...\n\n\n\n", MCPMessageType.PLAN_BEGIN) + yield self._create_output(MCP_GENERATE["START"][language], MCPMessageType.PLAN_BEGIN) # 选择工具并生成计划 selector = MCPSelector() top_tool = await selector.select_top_tool(self._call_vars.question, self.mcp_list) - planner = MCPPlanner(self._call_vars.question) + planner = MCPPlanner(self._call_vars.question, language) self._plan = await planner.create_plan(top_tool, self.max_steps) # 输出计划 @@ -94,12 +130,14 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): plan_str += f"[+] {plan_item.content}； {plan_item.tool}[{plan_item.instruction}]\n\n" yield self._create_output( - f"[MCP] 计划生成完成：\n\n{plan_str}\n\n\n\n", + MCP_GENERATE["END"][language].format(plan_str=plan_str), MCPMessageType.PLAN_END, data=self._plan.model_dump(), ) - async def _execute_plan_item(self, plan_item: MCPPlanItem) -> AsyncGenerator[CallOutputChunk, None]: + async def _execute_plan_item( + self, plan_item: MCPPlanItem, language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """执行单个计划项""" # 判断是否为Final if plan_item.tool == "Final": @@ -120,7 +158,7 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): # 调用工具 try: - result = await self._host.call_tool(tool, plan_item) + result = await self._host.call_tool(tool, plan_item, language) except Exception as e: err = f"[MCP] 工具 {tool.name} 调用失败: {e!s}" logger.exception(err) @@ -136,21 +174,21 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): }, ) - async def _generate_answer(self) -> AsyncGenerator[CallOutputChunk, None]: + async def _generate_answer(self, language) -> AsyncGenerator[CallOutputChunk, None]: """生成总结""" # 提示开始总结 yield self._create_output( - "[MCP] 正在总结任务结果...\n\n", + MCP_SUMMARY["START"][language], MCPMessageType.FINISH_BEGIN, ) # 生成答案 - planner = MCPPlanner(self._call_vars.question) + planner = MCPPlanner(self._call_vars.question, language) answer = await planner.generate_answer(self._plan, await self._host.assemble_memory()) # 输出结果 yield self._create_output( - f"[MCP] 任务完成\n\n---\n\n{answer}\n\n", + MCP_SUMMARY["END"][language].format(answer=answer), MCPMessageType.FINISH_END, data=MCPOutput( message=answer, @@ -166,8 +204,11 @@ class MCP(CoreCall, input_model=MCPInput, output_model=MCPOutput): """创建输出""" if self.text_output: return CallOutputChunk(type=CallOutputType.TEXT, content=text) - return CallOutputChunk(type=CallOutputType.DATA, content=MCPMessage( - msg_type=msg_type, - message=text.strip(), - data=data or {}, - ).model_dump_json()) + return CallOutputChunk( + type=CallOutputType.DATA, + content=MCPMessage( + msg_type=msg_type, + message=text.strip(), + data=data or {}, + ).model_dump_json(), + ) diff --git a/apps/scheduler/call/rag/rag.py b/apps/scheduler/call/rag/rag.py index e27327d8ad4d01387eeeb4f1644c056d32bc0dbe..f1f1bf7618b3a1aaabac872e9b2203428cefe136 100644 --- a/apps/scheduler/call/rag/rag.py +++ b/apps/scheduler/call/rag/rag.py @@ -3,7 +3,7 @@ import logging from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, ClassVar import httpx from fastapi import status @@ -13,7 +13,7 @@ from apps.common.config import Config from apps.llm.patterns.rewrite import QuestionRewrite from apps.scheduler.call.core import CoreCall from apps.scheduler.call.rag.schema import RAGInput, RAGOutput, SearchMethod -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -37,10 +37,27 @@ class RAG(CoreCall, input_model=RAGInput, output_model=RAGOutput): is_compress: bool = Field(description="是否压缩", default=False) tokens_limit: int = Field(description="token限制", default=8192) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "知识库", + "description": "查询知识库，从文档中获取必要信息", + }, + LanguageType.ENGLISH: { + "name": "Knowledge Base", + "description": "Query the knowledge base and obtain necessary information from documents", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="知识库", description="查询知识库，从文档中获取必要信息") + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> RAGInput: """初始化RAG工具""" @@ -58,7 +75,9 @@ class RAG(CoreCall, input_model=RAGInput, output_model=RAGOutput): tokensLimit=self.tokens_limit, ) - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """调用RAG工具""" data = RAGInput(**input_data) question_obj = QuestionRewrite() diff --git a/apps/scheduler/call/search/search.py b/apps/scheduler/call/search/search.py index 73d21d7b9956a19b6eaaf23467b4286d7fbb3d79..83c2c53f1d9d53bd73122e64426da3ade71faf7a 100644 --- a/apps/scheduler/call/search/search.py +++ b/apps/scheduler/call/search/search.py @@ -1,10 +1,11 @@ """搜索工具""" from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, ClassVar from apps.scheduler.call.core import CoreCall from apps.scheduler.call.search.schema import SearchInput, SearchOutput +from apps.schemas.enum_var import LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -16,11 +17,27 @@ from apps.schemas.scheduler import ( class Search(CoreCall, input_model=SearchInput, output_model=SearchOutput): """搜索工具""" + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "搜索", + "description": "获取搜索引擎的结果。", + }, + LanguageType.ENGLISH: { + "name": "Search", + "description": "Get the results of the search engine.", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="搜索", description="获取搜索引擎的结果") - + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> SearchInput: """初始化工具""" diff --git a/apps/scheduler/call/slot/prompt.py b/apps/scheduler/call/slot/prompt.py index e5650a4c5764a4ab739c6b66ad00838597c134a3..8a3f7ae8535f3420f05ca66b02d02429f143ca67 100644 --- a/apps/scheduler/call/slot/prompt.py +++ b/apps/scheduler/call/slot/prompt.py @@ -1,7 +1,9 @@ # Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. """自动参数填充工具的提示词""" +from apps.schemas.enum_var import LanguageType -SLOT_GEN_PROMPT = r""" +SLOT_GEN_PROMPT:dict[LanguageType, str] = { + LanguageType.CHINESE: r""" 你是一个可以使用工具的AI助手，正尝试使用工具来完成任务。 目前，你正在生成一个JSON参数对象，以作为调用工具的输入。 @@ -85,4 +87,91 @@ SLOT_GEN_PROMPT = r""" {{schema}} - """ + """, + LanguageType.ENGLISH: r""" + + You are an AI assistant capable of using tools to complete tasks. + Currently, you are generating a JSON parameter object as input for calling a tool. + Please generate a compliant JSON object based on user input, background information, tool information, and JSON Schema content. + + Background information will be provided in , tool information in , JSON Schema in , \ + and the user's question in . + Output the generated JSON object in . + + Requirements: + 1. Strictly follow the JSON format described in the JSON Schema. Do not fabricate non-existent fields. + 2. Prioritize using values from user input for JSON fields. If not available, use content from background information. + 3. Only output the JSON object. Do not include any explanations or additional content. + 4. Optional fields in the JSON Schema may be omitted. + 5. Examples are for illustration only. Do not copy content from examples or use them as output. + 6. Respond in the same language as the user's question by default, unless explicitly requested otherwise. + + + + + User asked about today's weather in Hangzhou. AI replied it's sunny, 20℃. User then asks about tomorrow's weather in Hangzhou. + + + What's the weather like in Hangzhou tomorrow? + + + Tool name: check_weather + Tool description: Query weather information for specified cities + + + { + "type": "object", + "properties": { + "city": { + "type": "string", + "description": "City name" + }, + "date": { + "type": "string", + "description": "Query date" + }, + "required": ["city", "date"] + } + } + + + { + "city": "Hangzhou", + "date": "tomorrow" + } + + + + + Historical summary of tasks given by user, provided in : + + {{summary}} + + Additional itemized information: + {{ facts }} + + + During this task, you have called some tools and obtained their outputs, provided in : + + {% for tool in history_data %} + + {{ tool.step_name }} + {{ tool.step_description }} + {{ tool.output_data }} + + {% endfor %} + + + + {{question}} + + + Tool name: {{current_tool["name"]}} + Tool description: {{current_tool["description"]}} + + + {{schema}} + + + """, +} diff --git a/apps/scheduler/call/slot/slot.py b/apps/scheduler/call/slot/slot.py index d24e1661e9dd7facc4bf590b24409052b6ec9104..cdbc543475e43ed74867aaf14ad046e88b59672b 100644 --- a/apps/scheduler/call/slot/slot.py +++ b/apps/scheduler/call/slot/slot.py @@ -3,7 +3,7 @@ import json from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, ClassVar from jinja2 import BaseLoader from jinja2.sandbox import SandboxedEnvironment @@ -15,7 +15,7 @@ from apps.scheduler.call.core import CoreCall from apps.scheduler.call.slot.prompt import SLOT_GEN_PROMPT from apps.scheduler.call.slot.schema import SlotInput, SlotOutput from apps.scheduler.slot.slot import Slot as SlotProcessor -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.pool import NodePool from apps.schemas.scheduler import CallInfo, CallOutputChunk, CallVars @@ -32,12 +32,31 @@ class Slot(CoreCall, input_model=SlotInput, output_model=SlotOutput): facts: list[str] = Field(description="事实信息", default=[]) step_num: int = Field(description="历史步骤数", default=1) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "参数自动填充", + "description": "根据步骤历史，自动填充参数", + }, + LanguageType.ENGLISH: { + "name": "Parameter Auto-Fill", + "description": "Auto-fill parameters based on step history.", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="参数自动填充", description="根据步骤历史，自动填充参数") - - async def _llm_slot_fill(self, remaining_schema: dict[str, Any]) -> tuple[str, dict[str, Any]]: + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) + + async def _llm_slot_fill( + self, remaining_schema: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> tuple[str, dict[str, Any]]: """使用大模型填充参数；若大模型解析度足够，则直接返回结果""" env = SandboxedEnvironment( loader=BaseLoader(), @@ -45,21 +64,24 @@ class Slot(CoreCall, input_model=SlotInput, output_model=SlotOutput): trim_blocks=True, lstrip_blocks=True, ) - template = env.from_string(SLOT_GEN_PROMPT) + template = env.from_string(SLOT_GEN_PROMPT[language]) conversation = [ {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": template.render( - current_tool={ - "name": self.name, - "description": self.description, - }, - schema=remaining_schema, - history_data=self._flow_history, - summary=self.summary, - question=self._question, - facts=self.facts, - )}, + { + "role": "user", + "content": template.render( + current_tool={ + "name": self.name, + "description": self.description, + }, + schema=remaining_schema, + history_data=self._flow_history, + summary=self.summary, + question=self._question, + facts=self.facts, + ), + }, ] # 使用大模型进行尝试 @@ -123,7 +145,9 @@ class Slot(CoreCall, input_model=SlotInput, output_model=SlotOutput): remaining_schema=remaining_schema, ) - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """执行参数填充""" data = SlotInput(**input_data) @@ -137,7 +161,7 @@ class Slot(CoreCall, input_model=SlotInput, output_model=SlotOutput): ).model_dump(by_alias=True, exclude_none=True), ) return - answer, slot_data = await self._llm_slot_fill(data.remaining_schema) + answer, slot_data = await self._llm_slot_fill(data.remaining_schema, language) slot_data = self._processor.convert_json(slot_data) remaining_schema = self._processor.check_json(slot_data) diff --git a/apps/scheduler/call/sql/sql.py b/apps/scheduler/call/sql/sql.py index 3e24301de508e06adf5cfdbf24b3d8ca37c0cc27..ca6ed90b8e7e6b0577c1fc34689a3b738e586f74 100644 --- a/apps/scheduler/call/sql/sql.py +++ b/apps/scheduler/call/sql/sql.py @@ -3,7 +3,7 @@ import logging from collections.abc import AsyncGenerator -from typing import Any +from typing import Any, ClassVar import httpx from fastapi import status @@ -12,7 +12,7 @@ from pydantic import Field from apps.common.config import Config from apps.scheduler.call.core import CoreCall from apps.scheduler.call.sql.schema import SQLInput, SQLOutput -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.scheduler import ( CallError, CallInfo, @@ -22,6 +22,17 @@ from apps.schemas.scheduler import ( logger = logging.getLogger(__name__) +MESSAGE = { + "invaild": { + LanguageType.CHINESE: "SQL查询错误：无法生成有效的SQL语句！", + LanguageType.ENGLISH: "SQL query error: Unable to generate valid SQL statements!", + }, + "fail": { + LanguageType.CHINESE: "SQL查询错误：SQL语句执行失败！", + LanguageType.ENGLISH: "SQL query error: SQL statement execution failed!", + }, +} + class SQL(CoreCall, input_model=SQLInput, output_model=SQLOutput): """SQL工具。用于调用外置的Chat2DB工具的API，获得SQL语句；再在PostgreSQL中执行SQL语句，获得数据。""" @@ -31,12 +42,27 @@ class SQL(CoreCall, input_model=SQLInput, output_model=SQLOutput): top_k: int = Field(description="生成SQL语句数量",default=5) use_llm_enhancements: bool = Field(description="是否使用大模型增强", default=False) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "SQL查询", + "description": "使用大模型生成SQL语句，用于查询数据库中的结构化数据", + }, + LanguageType.ENGLISH: { + "name": "SQL Query", + "description": "Use the foundation model to generate SQL statements to query structured data in the databases", + }, + } @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="SQL查询", description="使用大模型生成SQL语句，用于查询数据库中的结构化数据") + """ + 返回Call的名称和描述 + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) async def _init(self, call_vars: CallVars) -> SQLInput: """初始化SQL工具。""" @@ -113,16 +139,17 @@ class SQL(CoreCall, input_model=SQLInput, output_model=SQLOutput): return None, None - - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: - """运行SQL工具""" + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: + """运行SQL工具""" data = SQLInput(**input_data) # 生成SQL语句 sql_list = await self._generate_sql(data) if not sql_list: raise CallError( - message="SQL查询错误：无法生成有效的SQL语句！", + message=MESSAGE["invaild"][language], data={}, ) @@ -130,7 +157,7 @@ class SQL(CoreCall, input_model=SQLInput, output_model=SQLOutput): sql_exec_results, sql_exec = await self._execute_sql(sql_list) if sql_exec_results is None or sql_exec is None: raise CallError( - message="SQL查询错误：SQL语句执行失败！", + message=MESSAGE["fail"][language], data={}, ) diff --git a/apps/scheduler/call/suggest/prompt.py b/apps/scheduler/call/suggest/prompt.py index abc5e7d186500f917b4610b9c0e893dec7ef3c12..a9f61d7c1c5df6bc7fcee488d6ab05781b685d72 100644 --- a/apps/scheduler/call/suggest/prompt.py +++ b/apps/scheduler/call/suggest/prompt.py @@ -2,8 +2,11 @@ """问题推荐工具的提示词""" from textwrap import dedent +from apps.schemas.enum_var import LanguageType -SUGGEST_PROMPT = dedent(r""" +SUGGEST_PROMPT: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 根据提供的对话和附加信息（用户倾向、历史问题列表、工具信息等），生成三个预测问题。 @@ -89,4 +92,98 @@ SUGGEST_PROMPT = dedent(r""" 现在，进行问题生成： -""") +""" + ), + LanguageType.ENGLISH: dedent( + r""" + + + Generate three predicted questions based on the provided conversation and additional information (user preferences, historical question list, tool information, etc.). + The historical question list displays questions asked by the user before the historical conversation and is for background reference only. + The conversation will be given in the tag, the user preferences will be given in the tag, + the historical question list will be given in the tag, and the tool information will be given in the tag. + + Requirements for generating predicted questions: + + 1. Generate three predicted questions in the user's voice. They must be interrogative or imperative sentences and must be less than 30 words. + + 2. Predicted questions must be concise, without repetition, unnecessary information, or text other than the question. + + 3. Output must be in the following format: + + ```json + { + "predicted_questions": [ + "Predicted question 1", + "Predicted question 2", + "Predicted question 3" + ] + } + ``` + + + + What are the famous attractions in Hangzhou? + Hangzhou West Lake is a famous scenic spot in Hangzhou, Zhejiang Province, China, known for its beautiful natural scenery and rich cultural heritage. There are many famous attractions around West Lake, including the renowned Su Causeway, Bai Causeway, Broken Bridge, and the Three Pools Mirroring the Moon. West Lake is renowned for its clear waters and surrounding mountains, making it one of China's most famous lakes. + + + Briefly introduce Hangzhou + What are the famous attractions in Hangzhou? + + + Scenic Spot Search + Scenic Spot Information Search + + ["Hangzhou", "Tourism"] + + Now, generate questions: + + { + "predicted_questions": [ + "What is the ticket price for the West Lake Scenic Area in Hangzhou?", + "What are the famous attractions in Hangzhou?", + "What's the weather like in Hangzhou?" + ] + } + + + + Here's the actual data: + + + {% for message in conversation %} + <{{ message.role }}>{{ message.content }} + {% endfor %} + + + + {% if history %} + {% for question in history %} + {{ question }} + {% endfor %} + {% else %} + (No history question) + {% endif %} + + + + {% if tool %} + {{ tool.name }} + {{ tool.description }} + {% else %} + (No tool information) + {% endif %} + + + + {% if preference %} + {{ preference }} + {% else %} + (no user preference) + {% endif %} + + + Now, generate the question: + """ + ), +} diff --git a/apps/scheduler/call/suggest/suggest.py b/apps/scheduler/call/suggest/suggest.py index 1788fa0f4a8ede3af38264c9bb4a82628018086b..4ba912e68416a770e4864ba4635a15db1498d150 100644 --- a/apps/scheduler/call/suggest/suggest.py +++ b/apps/scheduler/call/suggest/suggest.py @@ -3,7 +3,7 @@ import random from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, ClassVar from jinja2 import BaseLoader from jinja2.sandbox import SandboxedEnvironment @@ -20,7 +20,7 @@ from apps.scheduler.call.suggest.schema import ( SuggestionInput, SuggestionOutput, ) -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.pool import NodePool from apps.schemas.record import RecordContent from apps.schemas.scheduler import ( @@ -47,11 +47,27 @@ class Suggestion(CoreCall, input_model=SuggestionInput, output_model=SuggestionO context: SkipJsonSchema[list[dict[str, str]]] = Field(description="Executor的上下文", exclude=True) conversation_id: SkipJsonSchema[str] = Field(description="对话ID", exclude=True) + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "问题推荐", + "description": "在答案下方显示推荐的下一个问题", + }, + LanguageType.ENGLISH: { + "name": "Question Suggestion", + "description": "Display the suggested next question under the answer", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="问题推荐", description="在答案下方显示推荐的下一个问题") + """ + 返回Call的名称和描述 + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) @classmethod async def instance(cls, executor: "StepExecutor", node: NodePool | None, **kwargs: Any) -> Self: @@ -124,8 +140,9 @@ class Suggestion(CoreCall, input_model=SuggestionInput, output_model=SuggestionO history_questions.append(record_data.question) return history_questions - - async def _exec(self, input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """运行问题推荐""" data = SuggestionInput(**input_data) @@ -141,7 +158,7 @@ class Suggestion(CoreCall, input_model=SuggestionInput, output_model=SuggestionO # 已推送问题数量 pushed_questions = 0 # 初始化Prompt - prompt_tpl = self._env.from_string(SUGGEST_PROMPT) + prompt_tpl = self._env.from_string(SUGGEST_PROMPT[language]) # 先处理configs for config in self.configs: diff --git a/apps/scheduler/call/summary/summary.py b/apps/scheduler/call/summary/summary.py index b605204e179246f915d561c0884fd277712faf33..c634880c035f22d5235b49d479cfa2ba31429047 100644 --- a/apps/scheduler/call/summary/summary.py +++ b/apps/scheduler/call/summary/summary.py @@ -2,14 +2,14 @@ """总结上下文工具""" from collections.abc import AsyncGenerator -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Self, ClassVar from pydantic import Field from apps.llm.patterns.executor import ExecutorSummary from apps.scheduler.call.core import CoreCall, DataBase from apps.scheduler.call.summary.schema import SummaryOutput -from apps.schemas.enum_var import CallOutputType +from apps.schemas.enum_var import CallOutputType, LanguageType from apps.schemas.pool import NodePool from apps.schemas.scheduler import ( CallInfo, @@ -28,10 +28,27 @@ class Summary(CoreCall, input_model=DataBase, output_model=SummaryOutput): context: ExecutorBackground = Field(description="对话上下文") + i18n_info: ClassVar[dict[str, dict]] = { + LanguageType.CHINESE: { + "name": "理解上下文", + "description": "使用大模型，理解对话上下文", + }, + LanguageType.ENGLISH: { + "name": "Context Understanding", + "description": "Use the foundation model to understand the conversation context", + }, + } + @classmethod def info(cls) -> CallInfo: - """返回Call的名称和描述""" - return CallInfo(name="理解上下文", description="使用大模型，理解对话上下文") + """ + 返回Call的名称和描述 + + :return: Call的名称和描述 + :rtype: CallInfo + """ + lang_info = cls.i18n_info.get(cls.language, cls.i18n_info[LanguageType.CHINESE]) + return CallInfo(name=lang_info["name"], description=lang_info["description"]) @classmethod async def instance(cls, executor: "StepExecutor", node: NodePool | None, **kwargs: Any) -> Self: @@ -51,20 +68,25 @@ class Summary(CoreCall, input_model=DataBase, output_model=SummaryOutput): """初始化工具，返回输入""" return DataBase() - - async def _exec(self, _input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def _exec( + self, _input_data: dict[str, Any], language: LanguageType = LanguageType.CHINESE + ) -> AsyncGenerator[CallOutputChunk, None]: """执行工具""" summary_obj = ExecutorSummary() - summary = await summary_obj.generate(background=self.context) + summary = await summary_obj.generate(background=self.context, language=language) self.tokens.input_tokens += summary_obj.input_tokens self.tokens.output_tokens += summary_obj.output_tokens yield CallOutputChunk(type=CallOutputType.TEXT, content=summary) - - async def exec(self, executor: "StepExecutor", input_data: dict[str, Any]) -> AsyncGenerator[CallOutputChunk, None]: + async def exec( + self, + executor: "StepExecutor", + input_data: dict[str, Any], + language: LanguageType = LanguageType.CHINESE, + ) -> AsyncGenerator[CallOutputChunk, None]: """执行工具""" - async for chunk in self._exec(input_data): + async for chunk in self._exec(input_data, language): content = chunk.content if not isinstance(content, str): err = "[SummaryCall] 工具输出格式错误" diff --git a/apps/scheduler/executor/agent.py b/apps/scheduler/executor/agent.py index fd44aacf740df6138598be3ecdbd26894600ccce..4114d76d1952df0a3462ff0d503b07c57ff989a1 100644 --- a/apps/scheduler/executor/agent.py +++ b/apps/scheduler/executor/agent.py @@ -93,14 +93,11 @@ class MCPAgentExecutor(BaseExecutor): self.tools[tool.id] = tool self.tool_list.extend(mcp_service.tools) self.tools[FINAL_TOOL_ID] = MCPTool( - id=FINAL_TOOL_ID, - name="Final Tool", - description="结束流程的工具", - mcp_id="", - input_schema={} + id=FINAL_TOOL_ID, name="Final Tool", description="结束流程的工具", mcp_id="", input_schema={}, + ) + self.tool_list.append( + MCPTool(id=FINAL_TOOL_ID, name="Final Tool", description="结束流程的工具", mcp_id="", input_schema={}), ) - self.tool_list.append(MCPTool(id=FINAL_TOOL_ID, name="Final Tool", - description="结束流程的工具", mcp_id="", input_schema={})) async def get_tool_input_param(self, is_first: bool) -> None: if is_first: diff --git a/apps/scheduler/mcp/host.py b/apps/scheduler/mcp/host.py index 8e11083900c20c29fcaf71b6535f3b442fc2313a..8e7b26e38e2eba24b9b9c735b93e6a5a12407970 100644 --- a/apps/scheduler/mcp/host.py +++ b/apps/scheduler/mcp/host.py @@ -14,7 +14,7 @@ from apps.llm.function import JsonGenerator from apps.scheduler.mcp.prompt import MEMORY_TEMPLATE from apps.scheduler.pool.mcp.client import MCPClient from apps.scheduler.pool.mcp.pool import MCPPool -from apps.schemas.enum_var import StepStatus +from apps.schemas.enum_var import StepStatus, LanguageType from apps.schemas.mcp import MCPPlanItem, MCPTool from apps.schemas.task import FlowStepHistory from apps.services.task import TaskManager @@ -25,10 +25,18 @@ logger = logging.getLogger(__name__) class MCPHost: """MCP宿主服务""" - def __init__(self, user_sub: str, task_id: str, runtime_id: str, runtime_name: str) -> None: + def __init__( + self, + user_sub: str, + task_id: str, + runtime_id: str, + runtime_name: str, + language: LanguageType = LanguageType.CHINESE, + ) -> None: """初始化MCP宿主""" self._user_sub = user_sub self._task_id = task_id + self.language = language # 注意：runtime在工作流中是flow_id和step_description，在Agent中可为标识Agent的id和description self._runtime_id = runtime_id self._runtime_name = runtime_name @@ -72,7 +80,7 @@ class MCPHost: continue context_list.append(context) - return self._env.from_string(MEMORY_TEMPLATE).render( + return self._env.from_string(MEMORY_TEMPLATE[self.language]).render( context_list=context_list, ) diff --git a/apps/scheduler/mcp/plan.py b/apps/scheduler/mcp/plan.py index cd4f5975eea3f023a92626966081c2d1eb33bdb7..78d695f2cc9fc47c995245f0e8cf059b60e76a3d 100644 --- a/apps/scheduler/mcp/plan.py +++ b/apps/scheduler/mcp/plan.py @@ -8,14 +8,16 @@ from apps.llm.function import JsonGenerator from apps.llm.reasoning import ReasoningLLM from apps.scheduler.mcp.prompt import CREATE_PLAN, FINAL_ANSWER from apps.schemas.mcp import MCPPlan, MCPTool +from apps.schemas.enum_var import LanguageType class MCPPlanner: """MCP 用户目标拆解与规划""" - def __init__(self, user_goal: str) -> None: + def __init__(self, user_goal: str, language: LanguageType = LanguageType.CHINESE) -> None: """初始化MCP规划器""" self.user_goal = user_goal + self.language = language self._env = SandboxedEnvironment( loader=BaseLoader, autoescape=True, @@ -25,7 +27,6 @@ class MCPPlanner: self.input_tokens = 0 self.output_tokens = 0 - async def create_plan(self, tool_list: list[MCPTool], max_steps: int = 6) -> MCPPlan: """规划下一步的执行流程，并输出""" # 获取推理结果 @@ -38,7 +39,7 @@ class MCPPlanner: async def _get_reasoning_plan(self, tool_list: list[MCPTool], max_steps: int) -> str: """获取推理大模型的结果""" # 格式化Prompt - template = self._env.from_string(CREATE_PLAN) + template = self._env.from_string(CREATE_PLAN[self.language]) prompt = template.render( goal=self.user_goal, tools=tool_list, @@ -88,7 +89,7 @@ class MCPPlanner: async def generate_answer(self, plan: MCPPlan, memory: str) -> str: """生成最终回答""" - template = self._env.from_string(FINAL_ANSWER) + template = self._env.from_string(FINAL_ANSWER[self.language]) prompt = template.render( plan=plan, memory=memory, diff --git a/apps/scheduler/mcp/prompt.py b/apps/scheduler/mcp/prompt.py index b906fb8f75dfafb2546aaef6eb36370645c57598..29721b31c92a84d875052f3097af6e3e5c6db82d 100644 --- a/apps/scheduler/mcp/prompt.py +++ b/apps/scheduler/mcp/prompt.py @@ -2,8 +2,11 @@ """MCP相关的大模型Prompt""" from textwrap import dedent +from apps.schemas.enum_var import LanguageType -MCP_SELECT = dedent(r""" +MCP_SELECT: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个乐于助人的智能助手。 你的任务是：根据当前目标，选择最合适的MCP Server。 @@ -61,8 +64,73 @@ MCP_SELECT = dedent(r""" ### 请一步一步思考： -""") -CREATE_PLAN = dedent(r""" +""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a helpful intelligent assistant. + Your task is to select the most appropriate MCP server based on your current goals. + + ## Things to note when selecting an MCP server: + + 1. Ensure you fully understand your current goals and select the most appropriate MCP server. + 2. Please select from the provided list of MCP servers; do not generate your own. + 3. Please provide the rationale for your choice before making your selection. + 4. Your current goals will be listed below, along with the list of MCP servers. + Please include your thought process in the "Thought Process" section and your selection in the "Selection Results" section. + 5. Your selection must be in JSON format, strictly following the template below. Do not output any additional content: + + ```json + { + "mcp": "The name of your selected MCP server" + } + ``` + + 6. The following example is for reference only. Do not use it as a basis for selecting an MCP server. + + ## Example + + ### Goal + + I need an MCP server to complete a task. + + ### MCP Server List + + - **mcp_1**: "MCP Server 1"; Description of MCP Server 1 + - **mcp_2**: "MCP Server 2"; Description of MCP Server 2 + + ### Think step by step: + + Because the current goal requires an MCP server to complete a task, select mcp_1. + + ### Select Result + + ```json + { + "mcp": "mcp_1" + } + ``` + + ## Let's get started! + + ### Goal + + {{goal}} + + ### MCP Server List + + {% for mcp in mcp_list %} + - **{{mcp.id}}**: "{{mcp.name}}"; {{mcp.description}} + {% endfor %} + + ### Think step by step: + +""" + ), +} +CREATE_PLAN: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划生成器。 请分析用户的目标，并生成一个计划。你后续将根据这个计划，一步一步地完成用户的目标。 @@ -94,8 +162,7 @@ CREATE_PLAN = dedent(r""" } ``` - - 在生成计划之前，请一步一步思考，解析用户的目标，并指导你接下来的生成。\ -思考过程应放置在 XML标签中。 + - 在生成计划之前，请一步一步思考，解析用户的目标，并指导你接下来的生成。思考过程应按步骤顺序放置在 XML标签中。 - 计划内容中，可以使用"Result[]"来引用之前计划步骤的结果。例如："Result[3]"表示引用第三条计划执行后的结果。 - 计划不得多于{{ max_num }}条，且每条计划内容应少于150字。 @@ -107,8 +174,7 @@ CREATE_PLAN = dedent(r""" {% for tool in tools %} - {{ tool.id }}{{tool.name}}；{{ tool.description }} {% endfor %} - - Final结束步骤，当执行到这一步时，\ -表示计划执行结束，所得到的结果将作为最终结果。 + - Final结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终结果。 # 样例 @@ -163,8 +229,114 @@ CREATE_PLAN = dedent(r""" {{goal}} # 计划 -""") -EVALUATE_PLAN = dedent(r""" +""" + ), + 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": "" + } + ] + } + ``` + + # Now start generating the plan: + + ## Goal + + {{goal}} + + # Plan +""" + ), +} +EVALUATE_PLAN: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划评估器。 请根据给定的计划，和当前计划执行的实际情况，分析当前计划是否合理和完整，并生成改进后的计划。 @@ -210,8 +382,61 @@ EVALUATE_PLAN = dedent(r""" # 现在开始评估计划： -""") -FINAL_ANSWER = dedent(r""" +""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan evaluator. + Based on the given plan and the actual execution of the current plan, analyze whether the current plan is reasonable and complete, and generate an improved plan. + + # 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 the completion of the plan execution. + + # Your previous plan was: + + {{ plan }} + + # The execution status of this plan is: + + The execution status of the plan will be placed in the XML tags. + + + {{ memory }} + + + # Notes when conducting the evaluation: + + - Please think step by step, analyze the user's goal, and guide your subsequent generation. The thinking process should be placed in the XML tags. + - The evaluation results are divided into two parts: + - Conclusion of the plan evaluation + - Improved plan + - Please output the evaluation results in the following JSON format: + + ```json + { + "evaluation": "Evaluation results", + "plans": [ + { + "content": "Improved plan content", + "tool": "Tool ID", + "instruction": "Tool instructions" + } + ] + } + ``` + + # Start evaluating the plan now: + +""" + ), +} +FINAL_ANSWER: dict[str, str] = { + LanguageType.CHINESE: dedent( + r""" 综合理解计划执行结果和背景信息，向用户报告目标的完成情况。 # 用户目标 @@ -230,12 +455,50 @@ FINAL_ANSWER = dedent(r""" # 现在，请根据以上信息，向用户报告目标的完成情况： -""") -MEMORY_TEMPLATE = dedent(r""" +""" + ), + LanguageType.ENGLISH: dedent( + r""" + Based on the understanding of the plan execution results and background information, report to the user the completion status of the goal. + + # User goal + + {{ goal }} + + # Plan execution status + + To achieve the above goal, you implemented the following plan: + + {{ memory }} + + # Other background information: + + {{ status }} + + # Now, based on the above information, please report to the user 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 %} -""") +""" + ), + LanguageType.ENGLISH: dedent( + r""" + {% for ctx in context_list %} + - Step {{ loop.index }}: {{ ctx.step_description }} + - Called tool `{{ ctx.step_id }}` and provided parameters `{{ ctx.input_data }}` + - Execution status: {{ ctx.status }} + - Got data: `{{ ctx.output_data }}` + {% endfor %} +""" + ), +} diff --git a/apps/scheduler/mcp/select.py b/apps/scheduler/mcp/select.py index f3d6e0d4ae3aea9508d48f33499ac2bcf3de98a9..e8b0e88c09ac1d1ac63f995b779da8483a2fce22 100644 --- a/apps/scheduler/mcp/select.py +++ b/apps/scheduler/mcp/select.py @@ -14,6 +14,7 @@ from apps.llm.reasoning import ReasoningLLM from apps.scheduler.mcp.prompt import ( MCP_SELECT, ) +from apps.schemas.enum_var import LanguageType from apps.schemas.mcp import ( MCPCollection, MCPSelectResult, @@ -48,10 +49,17 @@ class MCPSelector: logger.info("[MCPHelper] 查询MCP Server向量: %s, %s", query, mcp_list) mcp_table = await LanceDB().get_table("mcp") query_embedding = await Embedding.get_embedding([query]) - mcp_vecs = await (await mcp_table.search( - query=query_embedding, - vector_column_name="embedding", - )).where(f"id IN {MCPSelector._assemble_sql(mcp_list)}").limit(5).to_list() + mcp_vecs = ( + await ( + await mcp_table.search( + query=query_embedding, + vector_column_name="embedding", + ) + ) + .where(f"id IN {MCPSelector._assemble_sql(mcp_list)}") + .limit(5) + .to_list() + ) # 拿到名称和description logger.info("[MCPHelper] 查询MCP Server名称和描述: %s", mcp_vecs) @@ -72,10 +80,7 @@ class MCPSelector: return llm_mcp_list async def _get_mcp_by_llm( - self, - query: str, - mcp_list: list[dict[str, str]], - mcp_ids: list[str], + self, query: str, mcp_list: list[dict[str, str]], mcp_ids: list[str], language ) -> MCPSelectResult: """通过LLM选择最合适的MCP Server""" # 初始化jinja2环境 @@ -85,7 +90,7 @@ class MCPSelector: trim_blocks=True, lstrip_blocks=True, ) - template = env.from_string(MCP_SELECT) + template = env.from_string(MCP_SELECT[language]) # 渲染模板 mcp_prompt = template.render( mcp_list=mcp_list, @@ -133,9 +138,7 @@ class MCPSelector: return result async def select_top_mcp( - self, - query: str, - mcp_list: list[str], + self, query: str, mcp_list: list[str], language: LanguageType = LanguageType.CHINESE ) -> MCPSelectResult: """ 选择最合适的MCP Server @@ -146,10 +149,12 @@ class MCPSelector: llm_mcp_list = await self._get_top_mcp_by_embedding(query, mcp_list) # 通过LLM选择最合适的 - return await self._get_mcp_by_llm(query, llm_mcp_list, mcp_list) + return await self._get_mcp_by_llm(query, llm_mcp_list, mcp_list, language) @staticmethod - async def select_top_tool(query: str, mcp_list: list[str], top_n: int = 10) -> list[MCPTool]: + async def select_top_tool( + query: str, mcp_list: list[str], top_n: int = 10 + ) -> list[MCPTool]: """选择最合适的工具""" tool_vector = await LanceDB().get_table("mcp_tool") query_embedding = await Embedding.get_embedding([query]) diff --git a/apps/scheduler/mcp_agent/prompt.py b/apps/scheduler/mcp_agent/prompt.py index 824ece8a35dfb3b02afb0e5bd5602d6f5d0d1342..59abd383858e386f32fe7d0ec214cfbbf39b0996 100644 --- a/apps/scheduler/mcp_agent/prompt.py +++ b/apps/scheduler/mcp_agent/prompt.py @@ -1,9 +1,11 @@ # Copyright (c) Huawei Technologies Co., Ltd. 2023-2025. All rights reserved. """MCP相关的大模型Prompt""" - +from apps.schemas.enum_var import LanguageType from textwrap import dedent -MCP_SELECT = dedent(r""" +MCP_SELECT: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个乐于助人的智能助手。 你的任务是：根据当前目标，选择最合适的MCP Server。 @@ -61,16 +63,78 @@ MCP_SELECT = dedent(r""" ### 请一步一步思考： -""") -TOOL_SELECT = dedent(r""" + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are an intelligent assistant who is willing to help. + Your task is: according to the current goal, select the most suitable MCP Server. + + ## Notes when selecting MCP Server: + + 1. Make sure to fully understand the current goal and select the most suitable MCP Server. + 2. Please select from the given MCP Server list, do not generate MCP Server by yourself. + 3. Please first give your reason for selection, then give your selection. + 4. The current goal will be given below, and the MCP Server list will also be given below. + Please put your thinking process in the "Thinking Process" part, and put your selection in the "Selection Result" part. + 5. The selection must be in JSON format, strictly follow the template below, do not output any other content: + + ```json + { + "mcp": "The name of the MCP Server you selected" + } + ``` + 6. The example below is for reference only, do not use the content in the example as the basis for selecting MCP Server. + + ## Example + + ### Goal + + I need an MCP Server to complete a task. + + ### MCP Server List + + - **mcp_1**: "MCP Server 1"；Description of MCP Server 1 + - **mcp_2**: "MCP Server 2"；Description of MCP Server 2 + + ### Please think step by step: + + Because the current goal needs an MCP Server to complete a task, so select mcp_1. + + ### Selection Result + + ```json + { + "mcp": "mcp_1" + } + ``` + + ## Now start! + ### Goal + + {{goal}} + + ### MCP Server List + + {% for mcp in mcp_list %} + - **{{mcp.id}}**: "{{mcp.name}}"；{{mcp.description}} + {% endfor %} + + ### Please think step by step: + """ + ), +} +TOOL_SELECT: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个乐于助人的智能助手。 你的任务是：根据当前目标，附加信息，选择最合适的MCP工具。 ## 选择MCP工具时的注意事项： 1. 确保充分理解当前目标，选择实现目标所需的MCP工具。 - 2. 不要选择不存在的工具。 + 2. 请在给定的MCP工具列表中选择，不要自己生成MCP工具。 3. 可以选择一些辅助工具，但必须确保这些工具与当前目标相关。 4. 注意，返回的工具ID必须是MCP工具的ID，而不是名称。 - 5. 可以多选择一些工具，用于应对不同的情况。 + 5. 不要选择不存在的工具。 必须按照以下格式生成选择结果，不要输出任何其他内容： ```json { @@ -118,9 +182,69 @@ TOOL_SELECT = dedent(r""" {{additional_info}} # 输出 """ - ) + ), + LanguageType.ENGLISH: dedent( + r""" + You are an intelligent assistant who is willing to help. + Your task is: according to the current goal, additional information, select the most suitable MCP tool. + ## Notes when selecting MCP tool: + 1. Make sure to fully understand the current goal and select the MCP tool that can achieve the goal. + 2. Please select from the given MCP tool list, do not generate MCP tool by yourself. + 3. You can select some auxiliary tools, but you must ensure that these tools are related to the current goal. + 4. Note that the returned tool ID must be the ID of the MCP tool, not the name. + 5. Do not select non-existent tools. + Must generate the selection result in the following format, do not output any other content: + ```json + { + "tool_ids": ["tool_id1", "tool_id2", ...] + } + ``` -EVALUATE_GOAL = dedent(r""" + # Example + ## Goal + Optimize MySQL performance + ## MCP Tool List + + - mcp_tool_1 MySQL connection pool tool；used to optimize MySQL connection pool + - mcp_tool_2 MySQL performance tuning tool；used to analyze MySQL performance bottlenecks + - mcp_tool_3 MySQL query optimization tool；used to optimize MySQL query statements + - mcp_tool_4 MySQL index optimization tool；used to optimize MySQL index + - mcp_tool_5 File storage tool；used to store files + - mcp_tool_6 MongoDB tool；used to operate MongoDB database + + ## Additional Information + 1. The current MySQL database version is 8.0.26 + 2. The current MySQL database configuration file path is /etc/my.cnf, and contains the following configuration items + ```json + { + "max_connections": 1000, + "innodb_buffer_pool_size": "1G", + "query_cache_size": "64M" + } + ## Output + ```json + { + "tool_ids": ["mcp_tool_1", "mcp_tool_2", "mcp_tool_3", "mcp_tool_4"] + } + ``` + # Now start! + ## Goal + {{goal}} + ## MCP Tool List + + {% for tool in tools %} + - {{tool.id}} {{tool.name}}；{{tool.description}} + {% endfor %} + + ## Additional Information + {{additional_info}} + # Output + """ + ), +} +EVALUATE_GOAL: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划评估器。 请根据用户的目标和当前的工具集合以及一些附加信息，判断基于当前的工具集合，是否能够完成用户的目标。 如果能够完成，请返回`true`，否则返回`false`。 @@ -170,8 +294,65 @@ EVALUATE_GOAL = dedent(r""" # 附加信息 {{additional_info}} -""") -GENERATE_FLOW_NAME = dedent(r""" + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan evaluator. + Please judge whether the current tool set can complete the user's goal based on the user's goal and the current tool set and some additional information. + If it can be completed, return `true`, otherwise return `false`. + The reasoning process must be clear and understandable, so that people can understand your judgment basis. + Must answer in the following format: + ```json + { + "can_complete": true/false, + "resoning": "Your reasoning process" + } + ``` + + # Example + # Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + + # Tool Set + You can access and use some tools, which will be given in the XML tag. + + - mysql_analyzer Analyze MySQL database performance + - performance_tuner Tune database performance + - Final End step, when executing this step, it means that the plan execution is over, and the result obtained will be the final result. + + + # Additional Information + 1. The current MySQL database version is 8.0.26 + 2. The current MySQL database configuration file path is /etc/my.cnf + + ## + ```json + { + "can_complete": true, + "resoning": "The current tool set contains mysql_analyzer and performance_tuner, which can complete the performance analysis and optimization of MySQL database, so the user's goal can be completed." + } + ``` + + # Goal + {{goal}} + + # Tool Set + + {% for tool in tools %} + - {{tool.id}} {{tool.name}}；{{tool.description}} + {% endfor %} + + + # Additional Information + {{additional_info}} + + """ + ), +} +GENERATE_FLOW_NAME: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个智能助手，你的任务是根据用户的目标，生成一个合适的流程名称。 # 生成流程名称时的注意事项： @@ -189,8 +370,33 @@ GENERATE_FLOW_NAME = dedent(r""" # 目标 {{goal}} # 输出 - """) -GET_REPLAN_START_STEP_INDEX = dedent(r""" + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are an intelligent assistant, your task is to generate a suitable flow name based on the user's goal. + + # Notes when generating flow names: + 1. The flow name should be concise and clear, accurately expressing the process of achieving the user's goal. + 2. The flow name should include key operations or steps, such as "scan", "analyze", "tune", etc. + 3. The flow name should avoid using overly complex or professional terms, so that users can understand. + 4. The flow name should be as short as possible, less than 20 characters or words. + 5. Only output the flow name, do not output other content. + # Example + # Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + # Output + Scan MySQL database and analyze performance bottlenecks, and optimize it. + # Now start generating the flow name: + # Goal + {{goal}} + # Output + """ + ), +} +GET_REPLAN_START_STEP_INDEX: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个智能助手，你的任务是根据用户的目标、报错信息和当前计划和历史，获取重新规划的步骤起始索引。 # 样例 @@ -206,7 +412,7 @@ GET_REPLAN_START_STEP_INDEX = dedent(r""" "step_id": "step_1", "content": "生成端口扫描命令", "tool": "command_generator", - "instruction": "生成端口扫描命令：扫描 + "instruction": "生成端口扫描命令：扫描" }, { "step_id": "step_2", @@ -252,9 +458,77 @@ GET_REPLAN_START_STEP_INDEX = dedent(r""" # 历史 {{history}} # 输出 - """) - -CREATE_PLAN = dedent(r""" + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are an intelligent assistant, your task is to get the starting index of the step to be replanned based on the user's goal, error message, and current plan and history. + + # Example + # Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + # Error message + An error occurred while executing the port scan command: `- bash: curl: command not found`. + # Current plan + ```json + { + "plans": [ + { + "step_id": "step_1", + "content": "Generate port scan command", + "tool": "command_generator", + "instruction": "Generate port scan command: scan" + }, + { + "step_id": "step_2", + "content": "Execute the command generated by Result[0]", + "tool": "command_executor", + "instruction": "Execute port scan command" + } + ] + } + # History + [ + { + id: "0", + task_id: "task_1", + flow_id: "flow_1", + flow_name: "MYSQL Performance Tuning", + flow_status: "RUNNING", + step_id: "step_1", + step_name: "Generate port scan command", + step_description: "Generate port scan command: scan the port of the current MySQL database", + step_status: "FAILED", + input_data: { + "command": "nmap -p 3306 + "target": "localhost" + }, + output_data: { + "error": "- bash: curl: command not found" + } + } + ] + # Output + { + "start_index": 0, + "reasoning": "The first step of the current plan failed, the error message shows that the curl command was not found, which may be because the curl tool was not installed. Therefore, it is necessary to replan from the first step." + } + # Now start getting the starting index of the step to be replanned: + # Goal + {{goal}} + # Error message + {{error_message}} + # Current plan + {{current_plan}} + # History + {{history}} + # Output + """ + ), +} +CREATE_PLAN: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划生成器。 请分析用户的目标，并生成一个计划。你后续将根据这个计划，一步一步地完成用户的目标。 @@ -263,9 +537,7 @@ CREATE_PLAN = dedent(r""" 1. 能够成功完成用户的目标 2. 计划中的每一个步骤必须且只能使用一个工具。 3. 计划中的步骤必须具有清晰和逻辑的步骤，没有冗余或不必要的步骤。 - 4. 不要选择不存在的工具。 - 5. 计划中的最后一步必须是Final工具，以确保计划执行结束。 - 6. 生成的计划必须要覆盖用户的目标，当然需要考虑一些意外情况，可以有一定的冗余步骤。 + 4. 计划中的最后一步必须是Final工具，以确保计划执行结束。 # 生成计划时的注意事项： @@ -318,7 +590,8 @@ CREATE_PLAN = dedent(r""" - 在后台运行 - 执行top命令 3. 需要先选择MCP Server, 然后生成Docker命令, 最后执行命令 - ```json + + ```json { "plans": [ { @@ -327,17 +600,17 @@ CREATE_PLAN = dedent(r""" "instruction": "需要一个支持Docker容器运行的MCP Server" }, { - "content": "使用第一步选择的MCP Server，生成Docker命令", + "content": "使用Result[0]中选择的MCP Server，生成Docker命令", "tool": "command_generator", "instruction": "生成Docker命令：在后台运行alpine:latest容器，挂载/root到/data，执行top命令" }, { - "content": "执行第二步生成的Docker命令", + "content": "在Result[0]的MCP Server上执行Result[1]生成的命令", "tool": "command_executor", "instruction": "执行Docker命令" }, { - "content": "任务执行完成，容器已在后台运行", + "content": "任务执行完成，容器已在后台运行，结果为Result[2]", "tool": "Final", "instruction": "" } @@ -352,8 +625,112 @@ CREATE_PLAN = dedent(r""" {{goal}} # 计划 -""") -RECREATE_PLAN = dedent(r""" +""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan builder. + Analyze the user's goals and generate a plan. You will then follow this plan, step by step, to achieve the user's goals. + + # A good plan should: + + 1. Be able to successfully achieve the user's goals. + 2. Each step in the plan must use only one tool. + 3. The steps in the plan must have clear and logical progression, without redundant or unnecessary steps. + 4. The last step in the plan must be the Final tool to ensure the plan execution is complete. + + # Things to note when generating a plan: + + - Each plan contains 3 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 + - The plan must be generated in the following format, and no additional data should be output: + + ```json + { + "plans": [ + { + "content": "Plan content", + "tool": "Tool ID", + "instruction": "Tool instruction" + } + ] + } + ``` + + - Before generating a plan, please think step by step, analyze the user's goals, and guide your subsequent generation. + The thinking process should be placed in the XML tags. + - In the plan content, you can use "Result[]" to reference the results of the previous plan step. For example: "Result[3]" refers to the result after the third plan is executed. + - There should be no more than {{max_num}} plans, and each plan content should be less than 150 words. + + # Tools + + You can access and use a number of tools, listed within the XML tags. + + + {% for tool in tools %} + - {{tool.id}} {{tool.name}}; {{tool.description}} + {% endfor %} + + + # Example + + # Goal + + Run a new alpine:latest container in the background, mount the host's /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 the 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 commands: run the alpine:latest container in the background, mount /root to /data, and execute the top command" + }, + { + "content": "In the MCP of Result[0] Execute the command generated by Result[1] on the server", + "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": "" + } + ] + } + ``` + + # Now start generating the plan: + + # Goal + + {{goal}} + + # Plan + """ + ), +} +RECREATE_PLAN: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划重建器。 请根据用户的目标、当前计划和运行报错，重新生成一个计划。 @@ -363,9 +740,8 @@ RECREATE_PLAN = dedent(r""" 2. 计划中的每一个步骤必须且只能使用一个工具。 3. 计划中的步骤必须具有清晰和逻辑的步骤，没有冗余或不必要的步骤。 4. 你的计划必须避免之前的错误，并且能够成功执行。 - 5. 不要选择不存在的工具。 - 6. 计划中的最后一步必须是Final工具，以确保计划执行结束。 - 7. 生成的计划必须要覆盖用户的目标，当然需要考虑一些意外情况，可以有一定的冗余步骤。 + 5. 计划中的最后一步必须是Final工具，以确保计划执行结束。 + # 生成计划时的注意事项： - 每一条计划包含3个部分： @@ -501,74 +877,229 @@ RECREATE_PLAN = dedent(r""" {{error_message}} # 重新生成的计划 -""") -GEN_STEP = dedent(r""" - 你是一个计划生成器。 - 请根据用户的目标、当前计划和历史，生成一个新的步骤。 +""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan rebuilder. + Please regenerate a plan based on the user's goals, current plan, and runtime errors. - # 一个好的计划步骤应该： - 1.使用最适合的工具来完成当前步骤。 - 2.能够基于当前的计划和历史，完成阶段性的任务。 - 3.不要选择不存在的工具。 - 4.如果你认为当前已经达成了用户的目标，可以直接返回Final工具，表示计划执行结束。 + # A good plan should: + + 1. Successfully achieve the user's goals. + 2. Each step in the plan must use only one tool. + 3. The steps in the plan must have clear and logical progression, without redundant or unnecessary steps. + 4. Your plan must avoid previous errors and be able to be successfully executed. + 5. The last step in the plan must be the Final tool to ensure that the plan is complete. + + # Things to note when generating a plan: + + - Each plan contains 3 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 + - The plan must be generated in the following format, and no additional data should be output: - # 样例 1 - # 目标 - 我需要扫描当前mysql数据库，分析性能瓶颈, 并调优,我的ip是192.168.1.1，数据库端口是3306，用户名是root，密码是password - # 历史记录 - 第1步：生成端口扫描命令 - - 调用工具 `command_generator`，并提供参数 `帮我生成一个mysql端口扫描命令` - - 执行状态：成功 - - 得到数据：`{"command": "nmap -sS -p--open 192.168.1.1"}` - 第2步：执行端口扫描命令 - - 调用工具 `command_executor`，并提供参数 `{"command": "nmap -sS -p--open 192.168.1.1"}` - - 执行状态：成功 - - 得到数据：`{"result": "success"}` - # 工具 - - - mcp_tool_1 mysql_analyzer；用于分析数据库性能/description> - - mcp_tool_2 文件存储工具；用于存储文件 - - mcp_tool_3 mongoDB工具；用于操作MongoDB数据库 - - Final 结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终结果。 - - # 输出 ```json { - "tool_id": "mcp_tool_1", // 选择的工具ID - "description": "扫描ip为192.168.1.1的MySQL数据库，端口为3306，用户名为root，密码为password的数据库性能", + "plans": [ + { + "content": "Plan content", + "tool": "Tool ID", + "instruction": "Tool instruction" + } + ] } ``` - # 样例二 - # 目标 - 计划从杭州到北京的旅游计划 - # 历史记录 - 第1步：将杭州转换为经纬度坐标 - - 调用工具 `maps_geo_planner`，并提供参数 `{"city_from": "杭州", "address": "西湖"}` - - 执行状态：成功 - - 得到数据：`{"location": "123.456, 78.901"}` - 第2步：查询杭州的天气 - - 调用工具 `weather_query`，并提供参数 `{"location": "123.456, 78.901"}` - - 执行状态：成功 - - 得到数据：`{"weather": "晴", "temperature": "25°C"}` - 第3步：将北京转换为经纬度坐标 - - 调用工具 `maps_geo_planner`，并提供参数 `{"city_from": "北京", "address": "天安门"}` - - 执行状态：成功 - - 得到数据：`{"location": "123.456, 78.901"}` - 第4步：查询北京的天气 - - 调用工具 `weather_query`，并提供参数 `{"location": "123.456, 78.901"}` - - 执行状态：成功 - - 得到数据：`{"weather": "晴", "temperature": "25°C"}` - # 工具 + + - Before generating a plan, please think step by step, analyze the user's goals, and guide your subsequent generation. + The thinking process should be placed in the XML tags. + - In the plan content, you can use "Result[]" to reference the results of the previous plan step. For example: "Result[3]" refers to the result after the third plan is executed. + - There should be no more than {{max_num}} plans, and each plan content should be less than 150 words. + + # Objective + + Please scan the ports of the machine at 192.168.1.1 to see which ports are open. + # Tools + You can access and use a number of tools, which are listed within the XML tags. - - mcp_tool_4 maps_geo_planner；将详细的结构化地址转换为经纬度坐标。支持对地标性名胜景区、建筑物名称解析为经纬度坐标 - - mcp_tool_5 weather_query；天气查询，用于查询天气信息 - - mcp_tool_6 maps_direction_transit_integrated；根据用户起终点经纬度坐标规划综合各类公共（火车、公交、地铁）交通方式的通勤方案，并且返回通勤方案的数据，跨城场景下必须传起点城市与终点城市 - - Final Final；结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终结果。 - - # 输出 + - command_generator Generates command line instructions + - tool_selector Selects the appropriate tool + - command_executor Executes command line instructions + - Final This is the final step. When this step is reached, the plan execution ends, and the result is used as the final result. + # Current plan ```json { - "tool_id": "mcp_tool_6", // 选择的工具ID + "plans": [ + { + "content": "Generate port scan command", + "tool": "command_generator", + "instruction": "Generate port scan command: Scan open ports on 192.168.1.1" + }, + { + "content": "Execute the command generated in the first step", + "tool": "command_executor", + "instruction": "Execute the port scan command" + }, + { + "content": "Task execution completed", + "tool": "Final", + "instruction": "" + } + ] + } + ``` + # Run error + When executing the port scan command, an error occurred: `- bash: curl: command not found`. + # Regenerate the plan + + + 1. This goal requires a network scanning tool. First, select the appropriate network scanning tool. + 2. The goal can be broken down into the following parts: + - Generate the port scanning command + - Execute the port scanning command + 3. However, when executing the port scanning command, an error occurred: `- bash: curl: command not found`. + 4. I adjusted the plan to: + - Generate a command to check which network scanning tools the current machine supports + - Execute this command to check which network scanning tools the current machine supports + - Then select a network scanning tool + - Generate a port scanning command based on the selected network scanning tool + - Execute the port scanning command + ```json + { + "plans": [ + { + "content": "You need to generate a command to check which network scanning tools the current machine supports", + "tool": "command_generator", + "instruction": "Select which network scanning tools the current machine supports" + + }, + { + "content": "Execute the command generated in the first step to check which network scanning tools the current machine supports", + "tool": "command_executor", + "instruction": "Execute the command generated in the first step" + + }, + { + "content": "Select a network scanning tool from the results of the second step and generate a port scanning command", + "tool": "tool_selector", + "instruction": "Select a network scanning tool and generate a port scanning command" + + }, + { + "content": "Generate a port scan command based on the network scanning tool selected in step 3", + "tool": "command_generator", + "instruction": "Generate a port scan command: Scan the open ports on 192.168.1.1" + }, + { + "content": "Execute the port scan command generated in step 4", + "tool": "command_executor", + "instruction": "Execute the port scan command" + }, + { + "content": "Task execution completed", + "tool": "Final", + "instruction": "" + } + ] + } + ``` + + # Now start regenerating the plan: + + # Goal + + {{goal}} + + # Tools + + You can access and use a number of tools, which are listed within the XML tags. + + + {% for tool in tools %} + - {{tool.id}} {{tool.name}}; {{tool.description}} + {% endfor %} + + + # Current plan + {{current_plan}} + + # Run error + {{error_message}} + + # Regenerated plan + """ + ), +} +GEN_STEP: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" + 你是一个计划生成器。 + 请根据用户的目标、当前计划和历史，生成一个新的步骤。 + + # 一个好的计划步骤应该： + 1.使用最适合的工具来完成当前步骤。 + 2.能够基于当前的计划和历史，完成阶段性的任务。 + 3.不要选择不存在的工具。 + 4.如果你认为当前已经达成了用户的目标，可以直接返回Final工具，表示计划执行结束。 + + # 样例 1 + # 目标 + 我需要扫描当前mysql数据库，分析性能瓶颈, 并调优,我的ip是192.168.1.1，数据库端口是3306，用户名是root，密码是password + # 历史记录 + 第1步：生成端口扫描命令 + - 调用工具 `command_generator`，并提供参数 `帮我生成一个mysql端口扫描命令` + - 执行状态：成功 + - 得到数据：`{"command": "nmap -sS -p--open 192.168.1.1"}` + 第2步：执行端口扫描命令 + - 调用工具 `command_executor`，并提供参数 `{"command": "nmap -sS -p--open 192.168.1.1"}` + - 执行状态：成功 + - 得到数据：`{"result": "success"}` + # 工具 + + - mcp_tool_1 mysql_analyzer；用于分析数据库性能/description> + - mcp_tool_2 文件存储工具；用于存储文件 + - mcp_tool_3 mongoDB工具；用于操作MongoDB数据库 + - Final 结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终结果。 + + # 输出 + ```json + { + "tool_id": "mcp_tool_1", // 选择的工具ID + "description": "扫描ip为192.168.1.1的MySQL数据库，端口为3306，用户名为root，密码为password的数据库性能", + } + ``` + # 样例二 + # 目标 + 计划从杭州到北京的旅游计划 + # 历史记录 + 第1步：将杭州转换为经纬度坐标 + - 调用工具 `maps_geo_planner`，并提供参数 `{"city_from": "杭州", "address": "西湖"}` + - 执行状态：成功 + - 得到数据：`{"location": "123.456, 78.901"}` + 第2步：查询杭州的天气 + - 调用工具 `weather_query`，并提供参数 `{"location": "123.456, 78.901"}` + - 执行状态：成功 + - 得到数据：`{"weather": "晴", "temperature": "25°C"}` + 第3步：将北京转换为经纬度坐标 + - 调用工具 `maps_geo_planner`，并提供参数 `{"city_from": "北京", "address": "天安门"}` + - 执行状态：成功 + - 得到数据：`{"location": "123.456, 78.901"}` + 第4步：查询北京的天气 + - 调用工具 `weather_query`，并提供参数 `{"location": "123.456, 78.901"}` + - 执行状态：成功 + - 得到数据：`{"weather": "晴", "temperature": "25°C"}` + # 工具 + + - mcp_tool_4 maps_geo_planner；将详细的结构化地址转换为经纬度坐标。支持对地标性名胜景区、建筑物名称解析为经纬度坐标 + - mcp_tool_5 weather_query；天气查询，用于查询天气信息 + - mcp_tool_6 maps_direction_transit_integrated；根据用户起终点经纬度坐标规划综合各类公共（火车、公交、地铁）交通方式的通勤方案，并且返回通勤方案的数据，跨城场景下必须传起点城市与终点城市 + - Final Final；结束步骤，当执行到这一步时，表示计划执行结束，所得到的结果将作为最终结果。 + + # 输出 + ```json + { + "tool_id": "mcp_tool_6", // 选择的工具ID "description": "规划从杭州到北京的综合公共交通方式的通勤方案" } ``` @@ -583,9 +1114,97 @@ GEN_STEP = dedent(r""" - {{tool.id}} {{tool.name}}；{{tool.description}} {% endfor %} -""") +""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan generator. + Please generate a new step based on the user's goal, current plan, and history. + + # A good plan step should: + 1. Use the most appropriate tool for the current step. + 2. Complete the tasks at each stage based on the current plan and history. + 3. Do not select a tool that does not exist. + 4. If you believe the user's goal has been achieved, return to the Final tool to complete the plan execution. + + # Example 1 + # Objective + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. My IP address is 192.168.1.1, the database port is 3306, my username is root, and my password is password. + # History + Step 1: Generate a port scan command + - Call the `command_generator` tool and provide the `help me generate a MySQL port scan command` parameter. + - Execution status: Success. + - Result: `{"command": "nmap -sS -p --open 192.168.1.1"}` + Step 2: Execute the port scan command + - Call the `command_executor` tool and provide the `{"command": "nmap -sS -p --open 192.168.1.1"}` parameter. + - Execution status: Success. + - Result: `{"result": "success"}` + # Tools + + - mcp_tool_1 mysql_analyzer; used for analyzing database performance. + - mcp_tool_2 File storage tool; used for storing files. + - mcp_tool_3 MongoDB tool; used for operating MongoDB databases. + - Final This step completes the plan execution and the result is used as the final result. + + # Output + ```json + { + "tool_id": "mcp_tool_1", // Selected tool ID + "description": "Scan the database performance of the MySQL database with IP address 192.168.1.1, port 3306, username root, and password password", + } + ``` + # Example 2 + # Objective + Plan a trip from Hangzhou to Beijing + # History + Step 1: Convert Hangzhou to latitude and longitude coordinates + - Call the `maps_geo_planner` tool and provide `{"city_from": "Hangzhou", "address": "West Lake"}` + - Execution status: Success + - Result: `{"location": "123.456, 78.901"}` + Step 2: Query the weather in Hangzhou + - Call the `weather_query` tool and provide `{"location": "123.456, 78.901"}` + - Execution Status: Success + - Result: `{"weather": "Sunny", "temperature": "25°C"}` + Step 3: Convert Beijing to latitude and longitude coordinates + - Call the `maps_geo_planner` tool and provide `{"city_from": "Beijing", "address": "Tiananmen"}` + - Execution Status: Success + - Result: `{"location": "123.456, 78.901"}` + Step 4: Query the weather in Beijing + - Call the `weather_query` tool and provide `{"location": "123.456, 78.901"}` + - Execution Status: Success + - Result: `{"weather": "Sunny", "temperature": "25°C"}` + # Tools + + - mcp_tool_4 maps_geo_planner; Converts a detailed structured address into longitude and latitude coordinates. Supports parsing landmarks, scenic spots, and building names into longitude and latitude coordinates. + - mcp_tool_5 weather_query; Weather query, used to query weather information. + - mcp_tool_6 maps_direction_transit_integrated; Plans a commuting plan based on the user's starting and ending longitude and latitude coordinates, integrating various public transportation modes (train, bus, subway), and returns the commuting plan data. For cross-city scenarios, both the starting and ending cities must be provided. + - Final Final; Final step. When this step is reached, plan execution is complete, and the resulting result is used as the final result. + + # Output + ```json + { + "tool_id": "mcp_tool_6", // Selected tool ID + "description": "Plan a comprehensive public transportation commute from Hangzhou to Beijing" + } + ``` + # Now start generating steps: + # Goal + {{goal}} + # History + {{history}} + # Tools + + {% for tool in tools %} + - {{tool.id}} {{tool.name}}；{{tool.description}} + {% endfor %} + + """ + ), +} -TOOL_SKIP = dedent(r""" +TOOL_SKIP: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划执行器。 你的任务是根据当前的计划和用户目标，判断当前步骤是否需要跳过。 如果需要跳过，请返回`true`，否则返回`false`。 @@ -639,8 +1258,67 @@ TOOL_SKIP = dedent(r""" # 输出 """ - ) -RISK_EVALUATE = dedent(r""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan executor. + Your task is to determine whether the current step should be skipped based on the current plan and the user's goal. + If skipping is required, return `true`; otherwise, return `false`. + The answer must follow the following format: + ```json + { + "skip": true/false, + } + ``` + Note: + 1. Be cautious in your judgment and only decide whether to skip the current step when there is sufficient context in the historical messages. + # Example + # User Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + # History + Step 1: Generate a port scan command + - Call the `command_generator` tool with `{"command": "nmap -sS -p--open 192.168.1.1"}` + - Execution Status: Success + - Result: `{"command": "nmap -sS -p--open 192.168.1.1"}` + Step 2: Execute the port scan command + - Call the `command_executor` tool with `{"command": "nmap -sS -p--open 192.168.1.1"}` + - Execution Status: Success + - Result: `{"result": "success"}` + Step 3: Analyze the port scan results + - Call the `mysql_analyzer` tool with `{"host": "192.168.1.1", "port": 3306, "username": "root", "password": "password"}` + - Execution status: Success + - Result: `{"performance": "good", "bottleneck": "none"}` + # Current step + + step_4 + command_generator + Generate MySQL performance tuning commands + Generate MySQL performance tuning commands: Tune MySQL database performance + + # Output + ```json + { + "skip": true + } + ``` + # User goal + {{goal}} + # History + {{history}} + # Current step + + {{step_id}} + {{step_name}} + {{step_instruction}} + {{step_content}} + + # output + """ + ), +} +RISK_EVALUATE: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个工具执行计划评估器。 你的任务是根据当前工具的名称、描述和入参以及附加信息，判断当前工具执行的风险并输出提示。 ```json @@ -677,58 +1355,116 @@ RISK_EVALUATE = dedent(r""" } ``` # 工具 - < tool > - < name > {{tool_name}} < /name > - < description > {{tool_description}} < /description > - < / tool > + + {{tool_name}} + {{tool_description}} + # 工具入参 {{input_param}} # 附加信息 {{additional_info}} # 输出 """ - ) + ), + LanguageType.ENGLISH: dedent( + r""" + You are a tool execution plan evaluator. + Your task is to determine the risk of executing the current tool based on its name, description, input parameters, and additional information, and output a warning. + ```json + { + "risk": "low/medium/high", + "reason": "prompt message" + } + ``` + # Example + # Tool name + mysql_analyzer + # Tool description + Analyzes MySQL database performance + # Tool input + { + "host": "192.0.0.1", + "port": 3306, + "username": "root", + "password": "password" + } + # Additional information + 1. The current MySQL database version is 8.0.26 + 2. The current MySQL database configuration file path is /etc/my.cnf and contains the following configuration items + ```ini + [mysqld] + innodb_buffer_pool_size=1G + innodb_log_file_size=256M + ``` + # Output + ```json + { + "risk": "medium", + "reason": "This tool will connect to a MySQL database and analyze performance, which may impact database performance. This operation should only be performed in a non-production environment." + } + ``` + # Tool + + {{tool_name}} + {{tool_description}} + + # Tool Input Parameters + {{input_param}} + # Additional Information + {{additional_info}} + # Output + + """ + ), +} # 根据当前计划和报错信息决定下一步执行，具体计划有需要用户补充工具入参、重计划当前步骤、重计划接下来的所有计划 -TOOL_EXECUTE_ERROR_TYPE_ANALYSIS = dedent(r""" +TOOL_EXECUTE_ERROR_TYPE_ANALYSIS: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划决策器。 + 你的任务是根据用户目标、当前计划、当前使用的工具、工具入参和工具运行报错，决定下一步执行的操作。 请根据以下规则进行判断： 1. 仅通过补充工具入参来解决问题的，返回 missing_param; 2. 需要重计划当前步骤的，返回 decorrect_plan 3.推理过程必须清晰明了，能够让人理解你的判断依据，并且不超过100字。 你的输出要以json格式返回，格式如下： + ```json { "error_type": "missing_param/decorrect_plan, "reason": "你的推理过程" } ``` + # 样例 # 用户目标 我需要扫描当前mysql数据库，分析性能瓶颈, 并调优 # 当前计划 - {"plans": [ - { - "content": "生成端口扫描命令", - "tool": "command_generator", - "instruction": "生成端口扫描命令：扫描192.168.1.1的开放端口" - }, - { - "content": "在执行Result[0]生成的命令", - "tool": "command_executor", - "instruction": "执行端口扫描命令" - }, - { - "content": "任务执行完成，端口扫描结果为Result[2]", - "tool": "Final", - "instruction": "" - } - ]} + { + "plans": [ + { + "content": "生成端口扫描命令", + "tool": "command_generator", + "instruction": "生成端口扫描命令：扫描192.168.1.1的开放端口" + }, + { + "content": "在执行Result[0]生成的命令", + "tool": "command_executor", + "instruction": "执行端口扫描命令" + }, + { + "content": "任务执行完成，端口扫描结果为Result[2]", + "tool": "Final", + "instruction": "" + } + ] + } # 当前使用的工具 - < tool > - < name > command_executor < /name > - < description > 执行命令行指令 < /description > - < / tool > + + command_executor + 执行命令行指令 + # 工具入参 { "command": "nmap -sS -p--open 192.168.1.1" @@ -747,18 +1483,97 @@ TOOL_EXECUTE_ERROR_TYPE_ANALYSIS = dedent(r""" # 当前计划 {{current_plan}} # 当前使用的工具 - < tool > - < name > {{tool_name}} < /name > - < description > {{tool_description}} < /description > - < / tool > + + {{tool_name}} + {{tool_description}} + # 工具入参 {{input_param}} # 工具运行报错 {{error_message}} # 输出 """ - ) -IS_PARAM_ERROR = dedent(r""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan decider. + + Your task is to decide the next action based on the user's goal, the current plan, the tool being used, tool inputs, and tool errors. + Please make your decision based on the following rules: + 1. If the problem can be solved by simply adding tool inputs, return missing_param; + 2. If the current step needs to be replanned, return decorrect_plan. + 3. Your reasoning must be clear and concise, allowing the user to understand your decision. It should not exceed 100 words. + Your output should be returned in JSON format, as follows: + + ```json + { + "error_type": "missing_param/decorrect_plan, + "reason": "Your reasoning" + } + ``` + + # Example + # User Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + # Current Plan + { + "plans": [ + { + "content": "Generate port scan command", + "tool": "command_generator", + "instruction": "Generate port scan command: Scan the open ports of 192.168.1.1" + }, + { + "content": "Execute the command generated by Result[0]", + "tool": "command_executor", + "instruction": "Execute the port scan command" + }, + { + "content": "Task execution completed, the port scan result is Result[2]", + "tool": "Final", + "instruction": "" + } + ] + } + # Currently used tool + + command_executor + Execute command line instructions + + # Tool input parameters + { + "command": "nmap -sS -p--open 192.168.1.1" + } + # Tool running error + When executing the port scan command, an error occurred: `- bash: nmap: command not found`. + # Output + ```json + { + "error_type": "decorrect_plan", + "reason": "The second step of the current plan failed. The error message shows that the nmap command was not found. This may be because the nmap tool is not installed. Therefore, the current step needs to be replanned." + } + ``` + # User goal + {{goal}} + # Current plan + {{current_plan}} + # Currently used tool + + {{tool_name}} + {{tool_description}} + + # Tool input parameters + {{input_param}} + # Tool execution error + {{error_message}} + # Output + """ + ), +} + +IS_PARAM_ERROR: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个计划执行专家，你的任务是判断当前的步骤执行失败是否是因为参数错误导致的， 如果是，请返回`true`，否则返回`false`。 必须按照以下格式回答： @@ -817,9 +1632,74 @@ IS_PARAM_ERROR = dedent(r""" {{error_message}} # 输出 """ - ) + ), + LanguageType.ENGLISH: dedent( + r""" + You are a plan execution expert. Your task is to determine whether the current step execution failure is due to parameter errors. + If so, return `true`; otherwise, return `false`. + The answer must be in the following format: + ```json + { + "is_param_error": true/false, + } + ``` + # Example + # User Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. + # History + Step 1: Generate a port scan command + - Call the `command_generator` tool and provide `{"command": "nmap -sS -p--open 192.168.1.1"}` + - Execution Status: Success + - Result: `{"command": "nmap -sS -p--open 192.168.1.1"}` + Step 2: Execute the port scan command + - Call the `command_executor` tool and provide `{"command": "nmap -sS -p--open 192.168.1.1"}` + - Execution Status: Success + - Result: `{"result": "success"}` + # Current step + + step_3 + mysql_analyzer + Analyze MySQL database performance + + # Tool input parameters + { + "host": "192.0.0.1", + "port": 3306, + "username": "root", + "password": "password" + } + # Tool execution error + When executing the MySQL performance analysis command, an error occurred: `host is not correct`. + + # Output + ```json + { + "is_param_error": true + } + ``` + # User goal + {{goal}} + # History + {{history}} + # Current step + + {{step_id}} + {{step_name}} + {{step_instruction}} + + # Tool input parameters + {{input_param}} + # Tool error + {{error_message}} + # Output + """ + ), +} + # 将当前程序运行的报错转换为自然语言 -CHANGE_ERROR_MESSAGE_TO_DESCRIPTION = dedent(r""" +CHANGE_ERROR_MESSAGE_TO_DESCRIPTION: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个智能助手，你的任务是将当前程序运行的报错转换为自然语言描述。 请根据以下规则进行转换： 1. 将报错信息转换为自然语言描述，描述应该简洁明了，能够让人理解报错的原因和影响。 @@ -829,10 +1709,10 @@ CHANGE_ERROR_MESSAGE_TO_DESCRIPTION = dedent(r""" 5. 只输出自然语言描述，不要输出其他内容。 # 样例 # 工具信息 - < tool > - < name > port_scanner < /name > - < description > 扫描主机端口 < /description > - < input_schema > + + port_scanner + 扫描主机端口 + { "type": "object", "properties": { @@ -850,13 +1730,13 @@ CHANGE_ERROR_MESSAGE_TO_DESCRIPTION = dedent(r""" }, "password": { "type": "string", - "description": "密码" + "description": "密码" } }, "required": ["host", "port", "username", "password"] } - < /input_schema > - < / tool > + + # 工具入参 { "host": "192.0.0.1", @@ -870,21 +1750,91 @@ CHANGE_ERROR_MESSAGE_TO_DESCRIPTION = dedent(r""" 扫描端口时发生错误：密码不正确。请检查输入的密码是否正确，并重试。 # 现在开始转换报错信息： # 工具信息 - < tool > - < name > {{tool_name}} < /name > - < description > {{tool_description}} < /description > - < input_schema > + + {{tool_name}} + {{tool_description}} + {{input_schema}} - < /input_schema > - < / tool > + + # 工具入参 {{input_params}} # 报错信息 {{error_message}} # 输出 - """) + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are an intelligent assistant. Your task is to convert the error message generated by the current program into a natural language description. + Please follow the following rules for conversion: + 1. Convert the error message into a natural language description. The description should be concise and clear, allowing users to understand the cause and impact of the error. + 2. The description should include the specific content of the error and possible solutions. + 3. The description should avoid using overly technical terms so that users can understand it. + 4. The description should be as brief as possible, within 50 words. + 5. Only output the natural language description, do not output other content. + # Example + # Tool Information + + port_scanner + Scan host ports + + { + "type": "object", + "properties": { + "host": { + "type": "string", + "description": "Host address" + }, + "port": { + "type": "integer", + "description": "Port number" + }, + "username": { + "type": "string", + "description": "Username" + }, + "password": { + "type": "string", + "description": "Password" + } + }, + "required": ["host", "port", "username", "password"] + } + + + # Tool input + { + "host": "192.0.0.1", + "port": 3306, + "username": "root", + "password": "password" + } + # Error message + An error occurred while executing the port scan command: `password is not correct`. + # Output + An error occurred while scanning the port: The password is incorrect. Please check that the password you entered is correct and try again. + # Now start converting the error message: + # Tool information + + {{tool_name}} + {{tool_description}} + + {{input_schema}} + + + # Tool input parameters + {{input_params}} + # Error message + {{error_message}} + # Output + """ + ), +} # 获取缺失的参数的json结构体 -GET_MISSING_PARAMS = dedent(r""" +GET_MISSING_PARAMS: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个工具参数获取器。 你的任务是根据当前工具的名称、描述和入参和入参的schema以及运行报错，将当前缺失的参数设置为null，并输出一个JSON格式的字符串。 ```json @@ -909,42 +1859,126 @@ GET_MISSING_PARAMS = dedent(r""" } # 工具入参schema { - "type": "object", - "properties": { + "type": "object", + "properties": { + "host": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ], + "description": "MySQL数据库的主机地址（可以为字符串或null）" + }, + "port": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ], + "description": "MySQL数据库的端口号（可以是数字、字符串或null）" + }, + "username": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ], + "description": "MySQL数据库的用户名（可以为字符串或null）" + }, + "password": { + "anyOf": [ + {"type": "string"}, + {"type": "null"} + ], + "description": "MySQL数据库的密码（可以为字符串或null）" + } + }, + "required": ["host", "port", "username", "password"] + } + # 运行报错 + 执行端口扫描命令时，出现了错误：`password is not correct`。 + # 输出 + ```json + { + "host": "192.0.0.1", + "port": 3306, + "username": null, + "password": null + } + ``` + # 工具 + + {{tool_name}} + {{tool_description}} + + # 工具入参 + {{input_param}} + # 工具入参schema（部分字段允许为null） + {{input_schema}} + # 运行报错 + {{error_message}} + # 输出 + """ + ), + LanguageType.ENGLISH: dedent( + r""" + You are a tool parameter getter. + Your task is to set missing parameters to null based on the current tool's name, description, input parameters, input parameter schema, and runtime errors, and output a JSON-formatted string. + ```json + { + "host": "Please provide the host address", + "port": "Please provide the port number", + "username": "Please provide the username", + "password": "Please provide the password" + } + ``` + # Example + # Tool Name + mysql_analyzer + # Tool Description + Analyze MySQL database performance + # Tool Input Parameters + { + "host": "192.0.0.1", + "port": 3306, + "username": "root", + "password": "password" + } + # Tool Input Parameter Schema + { + "type": "object", + "properties": { "host": { "anyOf": [ {"type": "string"}, {"type": "null"} ], - "description": "MySQL数据库的主机地址（可以为字符串或null）" + "description": "MySQL database host address (can be a string or null)" }, "port": { "anyOf": [ {"type": "string"}, {"type": "null"} ], - "description": "MySQL数据库的端口号（可以是数字、字符串或null）" + "description": "MySQL database port number (can be a number, a string, or null)" }, "username": { "anyOf": [ {"type": "string"}, {"type": "null"} ], - "description": "MySQL数据库的用户名（可以为字符串或null）" + "description": "MySQL database username (can be a string or null)" }, "password": { "anyOf": [ - {"type": "string"}, - {"type": "null"} - ], - "description": "MySQL数据库的密码（可以为字符串或null）" - } - }, + {"type": "string"}, + {"type": "null"} + ], + "description": "MySQL database password (can be a string or null)" + } + }, "required": ["host", "port", "username", "password"] } - # 运行报错 - 执行端口扫描命令时，出现了错误：`password is not correct`。 - # 输出 + # Run error + When executing the port scan command, an error occurred: `password is not correct`. + # Output ```json { "host": "192.0.0.1", @@ -953,21 +1987,25 @@ GET_MISSING_PARAMS = dedent(r""" "password": null } ``` - # 工具 - < tool > - < name > {{tool_name}} < /name > - < description > {{tool_description}} < /description > - < / tool > - # 工具入参 + # Tool + + {{tool_name}} + {{tool_description}} + + # Tool input parameters {{input_param}} - # 工具入参schema（部分字段允许为null） + # Tool input parameter schema (some fields can be null) {{input_schema}} - # 运行报错 + # Run error {{error_message}} - # 输出 + # Output """ - ) -GEN_PARAMS = dedent(r""" + ), +} + +GEN_PARAMS: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个工具参数生成器。 你的任务是根据总的目标、阶段性的目标、工具信息、工具入参的schema和背景信息生成工具的入参。 注意： @@ -1041,21 +2079,100 @@ GEN_PARAMS = dedent(r""" {{background_info}} # 输出 """ - ) + ), + LanguageType.ENGLISH: dedent( + r""" + You are a tool parameter generator. + Your task is to generate tool input parameters based on the overall goal, phased goals, tool information, tool input parameter schema, and background information. + Note: + 1. The generated parameters must conform to the tool input parameter schema. + 2. The overall goal, phased goals, and background information must be fully understood and used to generate tool input parameters. + 3. The generated parameters must conform to the phased goals. + + # Example + # Tool Information + < tool > + < name >mysql_analyzer < /name > + < description > Analyze MySQL Database Performance < /description > + < / tool > + # Overall Goal + I need to scan the current MySQL database, analyze performance bottlenecks, and optimize it. The IP address is 192.168.1.1, the port is 3306, the username is root, and the password is password. + # Current Phase Goal + I need to connect to the MySQL database, analyze performance bottlenecks, and optimize it. # Tool input schema + { + "type": "object", + "properties": { + "host": { + "type": "string", + "description": "MySQL database host address" + }, + "port": { + "type": "integer", + "description": "MySQL database port number" + }, + "username": { + "type": "string", + "description": "MySQL database username" + }, + "password": { + "type": "string", + "description": "MySQL database password" + } + }, + "required": ["host", "port", "username", "password"] + } + # Background information + Step 1: Generate a port scan command + - Call the `command_generator` tool and provide the `Help me generate a MySQL port scan command` parameter + - Execution status: Success + - Received data: `{"command": "nmap -sS -p --open 192.168.1.1"}` + + Step 2: Execute the port scan command + - Call the `command_executor` tool and provide the parameters `{"command": "nmap -sS -p --open 192.168.1.1"}` + - Execution status: Success + - Received data: `{"result": "success"}` + # Output + ```json + { + "host": "192.168.1.1", + "port": 3306, + "username": "root", + "password": "password" + } + ``` + # Tool + < tool > + < name > {{tool_name}} < /name > + < description > {{tool_description}} < /description > + < / tool > + # Overall goal + {{goal}} + # Current stage goal + {{current_goal}} + # Tool input scheme + {{input_schema}} + # Background information + {{background_info}} + # Output + """ + ), +} -REPAIR_PARAMS = dedent(r""" +REPAIR_PARAMS: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 你是一个工具参数修复器。 你的任务是根据当前的工具信息、目标、工具入参的schema、工具当前的入参、工具的报错、补充的参数和补充的参数描述，修复当前工具的入参。 - + 注意： 1.最终修复的参数要符合目标和工具入参的schema。 - + # 样例 # 工具信息 - < tool > - < name > mysql_analyzer < /name > - < description > 分析MySQL数据库性能 < /description > - < / tool > + + mysql_analyzer + 分析MySQL数据库性能 + # 总目标 我需要扫描当前mysql数据库，分析性能瓶颈, 并调优 # 当前阶段目标 @@ -1109,10 +2226,10 @@ REPAIR_PARAMS = dedent(r""" } ``` # 工具 - < tool > - < name > {{tool_name}} < /name > - < description > {{tool_description}} < /description > - < / tool > + + {{tool_name}} + {{tool_description}} + # 总目标 {{goal}} # 当前阶段目标 @@ -1121,8 +2238,6 @@ REPAIR_PARAMS = dedent(r""" {{input_schema}} # 工具入参 {{input_param}} - # 工具描述 - {{tool_description}} # 运行报错 {{error_message}} # 补充的参数 @@ -1131,8 +2246,88 @@ REPAIR_PARAMS = dedent(r""" {{params_description}} # 输出 """ - ) -FINAL_ANSWER = dedent(r""" + ), + LanguageType.ENGLISH: dedent( + r""" + You are a tool parameter fixer. + Your task is to fix the current tool input parameters based on the current tool information, tool input parameter schema, tool current input parameters, tool error, supplemented parameters, and supplemented parameter descriptions. + + # Example + # Tool information + + mysql_analyzer + Analyze MySQL database performance + + # Tool input parameter schema + { + "type": "object", + "properties": { + "host": { + "type": "string", + "description": "MySQL database host address" + }, + "port": { + "type": "integer", + "description": "MySQL database port number" + }, + "username": { + "type": "string", + "description": "MySQL database username" + }, + "password": { + "type": "string", + "description": "MySQL database password" + } + }, + "required": ["host", "port", "username", "password"] + } + # Current tool input parameters + { + "host": "192.0.0.1", + "port": 3306, + "username": "root", + "password": "password" + } + # Tool error + When executing the port scan command, an error occurred: `password is not correct`. + # Supplementary parameters + { + "username": "admin", + "password": "admin123" + } + # Supplementary parameter description + The user wants to use the admin user and the admin123 password to connect to the MySQL database. + # Output + ```json + { + "host": "192.0.0.1", + "port": 3306, + "username": "admin", + "password": "admin123" + } + ``` + # Tool + + {{tool_name}} + {{tool_description}} + + # Tool input schema + {{input_schema}} + # Tool input parameters + {{input_param}} + # Runtime error + {{error_message}} + # Supplementary parameters + {{params}} + # Supplementary parameter descriptions + {{params_description}} + # Output + """ + ), +} +FINAL_ANSWER: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" 综合理解计划执行结果和背景信息，向用户报告目标的完成情况。 # 用户目标 @@ -1145,15 +2340,57 @@ FINAL_ANSWER = dedent(r""" {{memory}} + # 其他背景信息： + + {{status}} # 现在，请根据以上信息，向用户报告目标的完成情况： -""") -MEMORY_TEMPLATE = dedent(r""" - {% for ctx in context_list % } + """ + ), + LanguageType.ENGLISH: dedent( + r""" + Comprehensively understand the plan execution results and background information, and report the goal completion status to the user. + + # User Goal + + {{goal}} + + # Plan Execution Status + + To achieve the above goal, you implemented the following plan: + + {{memory}} + + # Additional Background Information: + + {{status}} + + # Now, based on the above information, report the goal completion status to the user: + + """ + ), +} + +MEMORY_TEMPLATE: dict[LanguageType, str] = { + LanguageType.CHINESE: dedent( + r""" + {% for ctx in context_list %} - 第{{loop.index}}步：{{ctx.step_description}} - - 调用工具 `{{ctx.step_id}}`，并提供参数 `{{ctx.input_data}}` - - 执行状态：{{ctx.step_status}} - - 得到数据：`{{ctx.output_data}}` - {% endfor % } -""") + - 调用工具 `{{ctx.step_id}}`，并提供参数 `{{ctx.input_data}}` + - 执行状态：{{ctx.status}} + - 得到数据：`{{ctx.output_data}}` + {% endfor %} + """ + ), + LanguageType.ENGLISH: dedent( + r""" + {% for ctx in context_list %} + - Step {{loop.index}}: {{ctx.step_description}} + - Call the tool `{{ctx.step_id}}` and provide the parameter `{{ctx.input_data}}` + - Execution status: {{ctx.status}} + - Receive data: `{{ctx.output_data}}` + {% endfor %} + """ + ), +}