diff --git a/backend/agents/create_agent_info.py b/backend/agents/create_agent_info.py
index 598ad2391..8158d652d 100644
--- a/backend/agents/create_agent_info.py
+++ b/backend/agents/create_agent_info.py
@@ -91,7 +91,7 @@ def _get_skills_for_template(
for s in enabled_skills
]
except Exception as e:
- logger.warning(f"Failed to get skills for template: {e}")
+ logger.error(f"Failed to get skills for agent {agent_id} (tenant={tenant_id}, version={version_no}): {e}", exc_info=True)
return []
@@ -457,6 +457,7 @@ async def create_agent_config(
# Build knowledge base summary
knowledge_base_summary = ""
+ kb_ids = []
try:
for tool in tool_list:
if "KnowledgeBaseSearchTool" == tool.class_name:
@@ -471,6 +472,7 @@ async def create_agent_config(
message = ElasticSearchService().get_summary(index_name=index_name)
summary = message.get("summary", "")
knowledge_base_summary += f"**{display_name}**: {summary}\n\n"
+ kb_ids.append(index_name)
except Exception as e:
logger.warning(
f"Failed to get summary for knowledge base {index_name}: {e}")
@@ -520,7 +522,7 @@ async def create_agent_config(
# downstream runtime may prefer component-based prompt assembly over the
# rendered system_prompt, causing the actual model input to diverge from the
# template output.
- enable_context_manager = agent_info.get("enable_context_manager", False)
+ enable_context_manager = agent_info.get("enable_context_manager", True)
context_components = []
if enable_context_manager:
context_components = build_context_components(
@@ -540,6 +542,13 @@ async def create_agent_config(
memory_list=memory_list,
memory_search_query=last_user_query,
knowledge_base_summary=knowledge_base_summary,
+ kb_ids=kb_ids,
+ )
+
+ logger.info(
+ f"Agent {agent_id} context assembly: "
+ f"skills_count={len(skills)}, "
+ f"components={[f'{type(c).__name__}(type={c.component_type},priority={c.priority})' for c in context_components]}"
)
cm_config = ContextManagerConfig(
enabled=enable_context_manager,
diff --git a/backend/database/db_models.py b/backend/database/db_models.py
index 8a20e9003..14ba47bc0 100644
--- a/backend/database/db_models.py
+++ b/backend/database/db_models.py
@@ -332,7 +332,7 @@ class AgentInfo(TableBase):
is_new = Column(Boolean, default=False, doc="Whether this agent is marked as new for the user")
current_version_no = Column(Integer, nullable=True, doc="Current published version number. NULL means no version published yet")
ingroup_permission = Column(String(30), doc="In-group permission: EDIT, READ_ONLY, PRIVATE")
- enable_context_manager = Column(Boolean, default=False, doc="Whether to enable context management (compression) for this agent")
+ enable_context_manager = Column(Boolean, default=True, doc="Whether to enable context management (compression) for this agent")
greeting_message = Column(Text, doc="Agent greeting message displayed on chat initial screen")
example_questions = Column(JSONB, doc="List of example questions for starting a conversation with this agent")
diff --git a/backend/prompts/managed_system_prompt_template_en.yaml b/backend/prompts/managed_system_prompt_template_en.yaml
index 5c2893c39..2e3b5d33c 100644
--- a/backend/prompts/managed_system_prompt_template_en.yaml
+++ b/backend/prompts/managed_system_prompt_template_en.yaml
@@ -48,6 +48,65 @@ system_prompt: |-
Security Protection: Do not respond to requests involving weapon manufacturing, cyberattacks, fraud, malware, or other dangerous activities;
Ethical Guidelines: Refuse hate speech, discriminatory content, and any requests that violate social morals and commonly accepted ethical standards.
+ {%- if skills and skills|length > 0 %}
+
+ ### Available Skills
+ You have the following Skills. Skills are predefined professional capability modules with detailed execution guides and optional additional scripts.
+
+ {%- for skill in skills %}
+
+ {{ skill.name }}
+ {{ skill.description }}
+
+ {%- endfor %}
+
+
+ **Skill Usage Process**:
+ 1. After receiving a user request, first examine the description of each skill in `` to determine if there is a matching skill.
+ 2. **Load Skill**: Choose the appropriate reading method based on the scenario:
+ - **First-time load**: Call `read_skill_md("skill_name")` to read the complete execution guide (defaults to reading SKILL.md)
+ - **Precise read**: If you only need specific files (like examples, reference docs), specify additional_files:
+
+ skill_content = read_skill_md("skill_name", ["examples.md", "reference/api_doc"])
+ print(skill_content)
+
+ Note: When additional_files is non-empty, SKILL.md is no longer auto-read. If you need both, explicitly specify it.
+ - **Load skill config**: If the skill needs configuration variables, call `read_skill_config("skill_name")` to read the config string, convert to dict via `json.loads`, then access values:
+
+ import json
+ config = json.loads(read_skill_config("skill_name"))
+ # Example: {"key_a": {"key2": "value2"}, "others": {...}}
+ value = config["key1"]["key2"]
+ print(value)
+
+ 3. **Follow Skill Guide**: After skill content is injected, strictly follow its steps. Do not skip steps or replace with your own code.
+ 4. **Execute Skill Script**: If the skill guide references additional scripts (like ``), call:
+
+ result = run_skill_script("skill_name", "script_path")
+ print(result)
+
+ For scripts needing extra params, pass them as a command-line string per the script's calling instructions.
+ Example for --param1 value1 --flag:
+
+ result = run_skill_script("skill_name", "script_path", "--param1 value1 --flag")
+ print(result)
+
+ Note: Only execute script paths explicitly declared in the skill guide. Never construct paths yourself.
+
+ 5. **Integrate Output**: Generate the final answer based on the skill guide's output format and script execution results.
+
+ 6. **Handle References**: When the skill content has reference markers or needs to reference other files, identify and call read_skill_md again:
+ - **Reference template recognition**: Look for patterns like `` or natural-language references ("see examples.md", "refer to reference/api_doc")
+ - **Auto-complete**: After discovering a reference, try reading the referenced file for more info
+ - **Example**:
+
+ # Skill content says "see examples.md for detailed examples"
+ additional_info = read_skill_md("skill_name", ["examples.md"])
+ print(additional_info)
+
+
+ {%- endif %}
+
### Execution Process
To solve tasks, you must plan forward through a series of steps in a loop of 'Think:' and 'Code:' sequences. **IMPORTANT: You must NOT output 'Observe Results:' before code execution. Observation results can ONLY be generated after code execution.**
@@ -124,6 +183,22 @@ system_prompt: |-
- No tools are currently available
{%- endif %}
+ {%- if skills and skills|length > 0 %}
+ - You have the skills listed in `` above. Scripts referenced in skills are called via the `run_skill_script()` function, which is provided by the platform and does not need to be imported.
+
+ ### Skill Usage Requirements
+ 1. **Skill First**: If a user request matches a skill's description, you must first call `read_skill_md()` to load the skill guide, then follow it. Do not skip the skill and write your own code to solve it.
+ 2. **Faithful Execution**: After reading the skill content, strictly follow the steps in the skill guide. Do not modify the process, skip steps, or replace the skill-defined workflow with generic code.
+ 3. **Script Calling Standards**: Only use the `run_skill_script` tool to execute scripts explicitly required by the skill guide. The `skill_name` and `script_path` passed in must exactly match the declarations in the skill guide. Do not construct or guess paths yourself. For scripts requiring additional parameters, pass the parameters as a command-line string to `run_skill_script`.
+ 4. **Failure Fallback**: If `read_skill_md` returns an error or `run_skill_script` fails, explain the situation to the user and try to provide an alternative using general reasoning.
+ 5. **Skill Composition**: If a task requires multiple skills working together, load and execute them in logical dependency order. The output of one skill can serve as the input for the next.
+
+
+ {%- else %}
+ - No skills are currently available
+ {%- endif %}
+
+
### Resource Usage Requirements
{{ constraint }}
diff --git a/backend/prompts/manager_system_prompt_template_en.yaml b/backend/prompts/manager_system_prompt_template_en.yaml
index 8ce58db29..655bf7cb2 100644
--- a/backend/prompts/manager_system_prompt_template_en.yaml
+++ b/backend/prompts/manager_system_prompt_template_en.yaml
@@ -48,6 +48,68 @@ system_prompt: |-
Security Protection: Do not respond to requests involving weapon manufacturing, cyberattacks, fraud, malware, or other dangerous activities;
Ethical Guidelines: Refuse hate speech, discriminatory content, and any requests that violate social morals and commonly accepted ethical standards.
+ {%- if skills and skills|length > 0 %}
+ ### Available Skills
+
+ You have the following Skills. Skills are predefined professional capability modules with detailed execution guides and optional additional scripts.
+
+
+ {%- for skill in skills %}
+
+ {{ skill.name }}
+ {{ skill.description }}
+
+ {%- endfor %}
+
+
+ **Skill Usage Process**:
+ 1. After receiving a user request, first examine the description of each skill in `` to determine if there is a matching skill.
+ 2. **Load Skill**: Choose the appropriate reading method based on the scenario:
+ - **First-time load**: Call `read_skill_md("skill_name")` to read the complete execution guide (defaults to reading SKILL.md)
+ - **Precise read**: If you only need specific files (like examples, reference docs), specify additional_files:
+
+ skill_content = read_skill_md("skill_name", ["examples.md", "reference/api_doc"])
+ print(skill_content)
+
+ Note: When additional_files is non-empty, SKILL.md is no longer auto-read. If you need both, explicitly specify it.
+
+ - **Load skill config**: If the skill needs configuration variables, call `read_skill_config("skill_name")` to read the config string, convert to dict via `json.loads`, then access values:
+
+ import json
+ config = json.loads(read_skill_config("skill_name"))
+ # Example: {"key_a": {"key2": "value2"}, "others": {...}}
+ value = config["key1"]["key2"]
+ print(value)
+
+
+ 3. **Follow Skill Guide**: After skill content is injected, strictly follow its steps. Do not skip steps or replace with your own code.
+
+ 4. **Execute Skill Script**: If the skill guide references additional scripts (like ``), call:
+
+ result = run_skill_script("skill_name", "script_path")
+ print(result)
+
+ For scripts needing extra params, pass them as a command-line string per the script's calling instructions.
+ Example for --param1 value1 --flag:
+
+ result = run_skill_script("skill_name", "script_path", "--param1 value1 --flag")
+ print(result)
+
+ Note: Only execute script paths explicitly declared in the skill guide. Never construct paths yourself.
+
+ 5. **Integrate Output**: Generate the final answer based on the skill guide's output format and script execution results.
+
+ 6. **Handle References**: When the skill content has reference markers or needs to reference other files, identify and call read_skill_md again:
+ - **Reference template recognition**: Look for patterns like `` or natural-language references ("see examples.md", "refer to reference/api_doc")
+ - **Auto-complete**: After discovering a reference, try reading the referenced file for more info
+ - **Example**:
+
+ # Skill content says "see examples.md for detailed examples"
+ additional_info = read_skill_md("skill_name", ["examples.md"])
+ print(additional_info)
+
+ {%- endif %}
+
### Execution Process
To solve tasks, you must plan forward through a series of steps in a loop of 'Think:' and 'Code:' sequences. **IMPORTANT: You must NOT output 'Observe Results:' before code execution. Observation results can ONLY be generated after code execution.**
@@ -164,7 +226,21 @@ system_prompt: |-
- No agents are currently available
{%- endif %}
- ### Resource Usage Requirements
+ 3. Skills
+ {%- if skills and skills|length > 0 %}
+ - You have the skills listed in `` above. Scripts referenced in skills are called via the `run_skill_script()` function, which is provided by the platform and does not need to be imported.
+
+ ### Skill Usage Requirements
+ 1. **Skill First**: If a user request matches a skill's description, you must first call `read_skill_md()` to load the skill guide, then follow it. Do not skip the skill and write your own code to solve it.
+ 2. **Faithful Execution**: After reading the skill content, strictly follow the steps in the skill guide. Do not modify the process, skip steps, or replace the skill-defined workflow with generic code.
+ 3. **Script Calling Standards**: Only use the `run_skill_script` tool to execute scripts explicitly required by the skill guide. The `skill_name` and `script_path` passed in must exactly match the declarations in the skill guide. Do not construct or guess paths yourself. For scripts requiring additional parameters, pass the parameters as a command-line string to `run_skill_script`.
+ 4. **Failure Fallback**: If `read_skill_md` returns an error or `run_skill_script` fails, explain the situation to the user and try to provide an alternative using general reasoning.
+ 5. **Skill Composition**: If a task requires multiple skills working together, load and execute them in logical dependency order. The output of one skill can serve as the input for the next.
+ {%- else %}
+ - No skills are currently available
+ {%- endif %}
+
+ ### Resource Usage Requirements
{{ constraint }}
### Python Code Specifications
diff --git a/backend/utils/context_utils.py b/backend/utils/context_utils.py
index 740bf66df..517cc3a46 100644
--- a/backend/utils/context_utils.py
+++ b/backend/utils/context_utils.py
@@ -266,7 +266,6 @@ def _format_skills_description(
def _format_tools_description(
tools: Dict[str, Any],
- knowledge_base_summary: Optional[str] = None,
language: str = "zh",
is_manager: bool = True,
) -> str:
@@ -279,10 +278,16 @@ def _format_tools_description(
"""
if not tools:
no_tools_msg = "- 当前没有可用的工具" if language == "zh" else "- No tools are currently available"
- return no_tools_msg
+ prefix = "1. 工具\n" if language == "zh" else "1. Tools\n"
+ return prefix + no_tools_msg
lines = []
+ if language == "zh":
+ lines.append("1. 工具")
+ else:
+ lines.append("1. Tools")
+
if language == "zh":
lines.append("- 你只能使用以下工具，不得使用任何其他工具：")
else:
@@ -320,15 +325,6 @@ def _format_tools_description(
lines.append(f" Accepts input: {inputs}")
lines.append(f" Returns output type: {output_type}")
- # Knowledge base summary
- if knowledge_base_summary:
- if language == "zh":
- lines.append("- knowledge_base_search工具只能使用以下知识库索引，请根据用户问题选择最相关的一个或多个知识库索引：")
- lines.append(f" {knowledge_base_summary}")
- else:
- lines.append("- knowledge_base_search tool can only use the following knowledge base indexes, please select the most relevant one or more knowledge base indexes based on the user's question:")
- lines.append(f" {knowledge_base_summary}")
-
# File URL usage guide
lines.append("")
if language == "zh":
@@ -375,6 +371,11 @@ def _format_managed_agents_description(
lines = []
+ if language == "zh":
+ lines.append("2. 助手")
+ else:
+ lines.append("2. Agents")
+
if language == "zh":
lines.append("你可以使用以下内部助手（通过函数调用方式协作）：")
for name, agent in managed_agents.items():
@@ -462,6 +463,7 @@ def _format_external_agents_description(
def _format_skills_usage_requirements(
skills: List[Dict[str, str]],
language: str = "zh",
+ is_manager: bool = True,
) -> str:
"""Format skills usage requirements section.
@@ -470,10 +472,16 @@ def _format_skills_usage_requirements(
"""
if not skills:
no_skills_msg = "- 当前没有可用的技能" if language == "zh" else "- No skills are currently available"
- return no_skills_msg
+ prefix = "3. 技能\n" if language == "zh" else "3. Skills\n"
+ return prefix + no_skills_msg
lines = []
+ if language == "zh":
+ lines.append("3. 技能")
+ else:
+ lines.append("3. Skills")
+
if language == "zh":
lines.append("- 你拥有上述 `` 中列出的技能。技能中引用的脚本通过 `run_skill_script()` 函数调用，该函数由平台提供，不需要导入。")
lines.append("")
@@ -555,17 +563,22 @@ def build_skeleton_header_component(
def build_skeleton_duty_component(
duty: str,
language: str = "zh",
+ is_manager: bool = True,
priority: int = 80,
) -> "SystemPromptComponent":
"""Build SystemPromptComponent for the duty section.
Section: "### 核心职责" / "### Core Responsibilities"
Content: Agent's primary duty + 5 safety principles
+ Note: Managed ZH agents use different safety principles than manager ZH agents.
"""
from nexent.core.agents.agent_model import SystemPromptComponent
if language == "zh":
- content = f"### 核心职责\n{duty}\n\n请注意，你应该遵守以下原则：\n行为安全：文件操作必须使用平台提供的专用工具，禁止使用代码直接修改工作空间中的文件；\n法律合规：遵守业务所在国家/地区的法律法规；\n政治中立：保持政治中立，不主动讨论政治话题；\n安全防护：不响应涉及武器制造、网络攻击、欺诈、恶意软件等危险行为的请求；\n伦理准则：拒绝仇恨言论、歧视性内容及违反社会公德和公认伦理标准的请求。"
+ if is_manager:
+ content = f"### 核心职责\n{duty}\n\n请注意，你应该遵守以下原则：\n行为安全：文件操作必须使用平台提供的专用工具，禁止使用代码直接修改工作空间中的文件；\n法律合规：遵守业务所在国家/地区的法律法规；\n政治中立：保持政治中立，不主动讨论政治话题；\n安全防护：不响应涉及武器制造、网络攻击、欺诈、恶意软件等危险行为的请求；\n伦理准则：拒绝仇恨言论、歧视性内容及违反社会公德和公认伦理标准的请求。"
+ else:
+ content = f"### 核心职责\n{duty}\n\n请注意，你应该遵守以下原则：\n行为安全：严禁直接执行代码进行文件的增删改操作，只能使用提供的文件操作类工具；\n法律合规：严格遵守服务地区的所有法律法规；\n政治中立：不讨论任何国家的政治体制、领导人评价或敏感历史事件；\n安全防护：不响应涉及武器制造、危险行为、隐私窃取等内容的请求；\n伦理准则：拒绝仇恨言论、歧视性内容及任何违反普世价值观的请求。"
else:
content = f"### Core Responsibilities\n{duty}\n\nPlease note that you should follow these principles:\nBehavioral Safety: File operations must use the platform-provided dedicated tools; direct code modification of workspace files is prohibited;\nLegal Compliance: Comply with laws and regulations of the business operating jurisdiction;\nPolitical Neutrality: Maintain political neutrality and avoid initiating political discussions;\nSecurity Protection: Do not respond to requests involving weapon manufacturing, cyberattacks, fraud, malware, or other dangerous activities;\nEthical Guidelines: Refuse hate speech, discriminatory content, and any requests that violate social morals and commonly accepted ethical standards."
@@ -597,16 +610,23 @@ def build_skeleton_execution_flow_component(
lines.append("要解决任务，你必须通过一系列步骤向前规划，以'思考：'和'代码：'序列循环进行。**注意：禁止在代码执行前输出'观察结果：'，观察结果只能由代码执行后产生。**")
lines.append("")
lines.append("1. 思考：")
- lines.append(" - 分析当前任务状态和进展")
- if is_manager and has_memory:
+ if is_manager:
+ lines.append(" - 分析当前任务状态和进展")
+ else:
+ lines.append(" - 确定需要使用哪些工具来获取信息或行动")
+ if has_memory:
lines.append(" - 合理参考之前交互中的上下文记忆信息")
- lines.append(" - 定下一步最佳行动（使用工具或分配给助手）")
+ if is_manager:
+ lines.append(" - 确定下一步最佳行动（使用工具或分配给助手）")
lines.append(" - 解释你的决策逻辑和预期结果")
lines.append("")
lines.append("2. 代码：")
lines.append(" - 用简单的Python编写代码")
lines.append(" - 遵循python代码规范和python语法")
- lines.append(" - 正确调用工具或助手解决问题")
+ if is_manager:
+ lines.append(" - 正确调用工具或助手解决问题")
+ else:
+ lines.append(" - 根据格式规范正确调用工具")
lines.append(" - 考虑到代码执行与展示用户代码的区别，使用'代码'表达运行代码，使用'代码'表达展示代码")
lines.append(" - 注意运行的代码不会被用户看到，所以如果用户需要看到代码，你需要使用'代码'表达展示代码。")
lines.append(" - **重要**：代码执行后，系统会返回 \"Observation:\" 标记的内容（这是真实的执行结果）。请基于这些真实结果继续下一步思考，**不要在代码执行前自行编造观察结果**。")
@@ -633,21 +653,31 @@ def build_skeleton_execution_flow_component(
lines.append(" - 避免在Markdown中使用HTML标签，优先使用Markdown原生语法")
lines.append(" - 代码块中的代码应保持原始格式，不要添加额外的转义字符")
lines.append(" - 若未使用检索工具，则不添加任何引用标记")
+ if not is_manager:
+ lines.append("")
+ lines.append("注意最后生成的回答要语义连贯，信息清晰，可读性高。")
else:
lines = ["### Execution Process"]
lines.append("To solve tasks, you must plan forward through a series of steps in a loop of 'Think:' and 'Code:' sequences. **IMPORTANT: You must NOT output 'Observe Results:' before code execution. Observation results can ONLY be generated after code execution.**")
lines.append("")
lines.append("1. Think:")
- lines.append(" - Analyze current task status and progress")
- if is_manager and has_memory:
+ if is_manager:
+ lines.append(" - Analyze current task status and progress")
+ else:
+ lines.append(" - Determine which tools need to be used to obtain information or take action")
+ if has_memory:
lines.append(" - Reference relevant contextual memories from previous interactions when applicable")
- lines.append(" - Determine the best next action (use tools or delegate to agents)")
+ if is_manager:
+ lines.append(" - Determine the best next action (use tools or delegate to agents)")
lines.append(" - Explain your decision logic and expected results")
lines.append("")
lines.append("2. Code:")
lines.append(" - Write code in simple Python")
lines.append(" - Follow Python coding standards and Python syntax")
- lines.append(" - Correctly call tools or agents to solve problems")
+ if is_manager:
+ lines.append(" - Correctly call tools or agents to solve problems")
+ else:
+ lines.append(" - Call tools correctly according to format specifications")
lines.append(" - To distinguish between code execution and displaying user code, use 'code' for executing code and 'code' for displaying code")
lines.append(" - Note that executed code is not visible to users. If users need to see the code, use 'code' for displaying code.")
lines.append(" - **IMPORTANT**: After code execution, the system will return content with \"Observation:\" marker (this is the real execution result). Please continue your next thinking based on these real results. **Do NOT fabricate observation results before code execution.**")
@@ -674,6 +704,9 @@ def build_skeleton_execution_flow_component(
lines.append(" - Avoid using HTML tags in Markdown, prioritize native Markdown syntax")
lines.append(" - Code in code blocks should maintain original format, do not add extra escape characters")
lines.append(" - If no retrieval tools are used, do not add any reference marks")
+ if not is_manager:
+ lines.append("")
+ lines.append("Note that the final generated answer should be semantically coherent, with clear information and high readability.")
content = "\n".join(lines)
@@ -782,6 +815,35 @@ def build_skeleton_footer_component(
)
+def build_available_resources_header_component(
+ is_manager: bool = True,
+ language: str = "zh",
+ priority: int = 55,
+) -> "SystemPromptComponent":
+ """Build SystemPromptComponent for the Available Resources section header.
+
+ Manager agents get a preamble restricting resources; managed agents get only the heading.
+ """
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ if language == "zh":
+ if is_manager:
+ content = "### 可用资源\n你只能使用以下资源，不得使用任何其他工具或助手："
+ else:
+ content = "### 可用资源"
+ else:
+ if is_manager:
+ content = "### Available Resources\nYou can only use the following resources, and may not use any other tools or agents:"
+ else:
+ content = "### Available Resources"
+
+ return SystemPromptComponent(
+ content=content,
+ template_name="available_resources_header",
+ priority=priority,
+ )
+
+
# =============================================================================
# SECTION 3: Piecewise component builders (existing, enhanced)
# =============================================================================
@@ -830,7 +892,6 @@ def build_tools_component(
formatted_desc = _format_tools_description(
tools,
- knowledge_base_summary=knowledge_base_summary,
language=language,
is_manager=is_manager,
)
@@ -913,6 +974,7 @@ def build_knowledge_base_component(
knowledge_base_summary: str,
kb_ids: Optional[List[str]] = None,
priority: int = 10,
+ language: str = "zh",
) -> "KnowledgeBaseComponent":
"""Build KnowledgeBaseComponent from knowledge base summary.
@@ -920,14 +982,24 @@ def build_knowledge_base_component(
knowledge_base_summary: Summary text from knowledge bases
kb_ids: List of knowledge base IDs used
priority: Component priority for selection
+ language: Language code ('zh' or 'en')
Returns:
KnowledgeBaseComponent instance
"""
from nexent.core.agents.agent_model import KnowledgeBaseComponent
+ if knowledge_base_summary:
+ if language == "zh":
+ guidance = "knowledge_base_search 工具只能使用以下知识库索引，请根据用户的问题选择最相关的一个或多个知识库索引：\n"
+ else:
+ guidance = "knowledge_base_search tool can only use the following knowledge base indexes, please select the most relevant one or more knowledge base indexes based on the user's question:\n"
+ prefixed_summary = guidance + knowledge_base_summary
+ else:
+ prefixed_summary = knowledge_base_summary
+
return KnowledgeBaseComponent(
- summary=knowledge_base_summary,
+ summary=prefixed_summary,
kb_ids=kb_ids or [],
priority=priority,
)
@@ -1046,9 +1118,10 @@ def build_system_prompt_component(
def build_skills_usage_component(
skills: List[Dict[str, str]],
language: str = "zh",
+ is_manager: bool = True,
priority: int = 40,
-) -> "SystemPromptComponent":
- """Build SystemPromptComponent for skills usage requirements.
+) -> "SkillsComponent":
+ """Build SkillsComponent for skills usage requirements.
This is a skeleton-like component but its content depends on
whether skills exist, so it's built dynamically.
@@ -1056,17 +1129,18 @@ def build_skills_usage_component(
Args:
skills: List of skill dicts
language: Language code ('zh' or 'en')
+ is_manager: Whether this is a manager agent
priority: Component priority
Returns:
- SystemPromptComponent instance
+ SkillsComponent instance
"""
- from nexent.core.agents.agent_model import SystemPromptComponent
+ from nexent.core.agents.agent_model import SkillsComponent
- content = _format_skills_usage_requirements(skills, language=language)
- return SystemPromptComponent(
- content=content,
- template_name="skills_usage",
+ content = _format_skills_usage_requirements(skills, language=language, is_manager=is_manager)
+ return SkillsComponent(
+ skills=skills,
+ formatted_description=content,
priority=priority,
)
@@ -1141,20 +1215,22 @@ def build_context_components(
Piecewise assembly: Each semantic section is emitted as a dedicated
ContextComponent, assembled in the exact order matching Jinja2 templates.
- Assembly order (12 sections):
+ Assembly order (15 sections):
1. Header (基本信息)
2. Memory (上下文记忆) - if memory_list exists
3. Duty (核心职责 + 安全准则)
4. Skills (可用技能 + 6步流程) - if skills exist
5. Execution Flow (执行流程 + 输出规范)
- 6. Tools (可用资源/1. 工具 + 文件链接指南)
- 7. Managed Agents (可用资源/2. 助手) - if managed_agents exist
- 8. External Agents (外部助手) - if external_a2a_agents exist
- 9. Agent Fallback (当前没有可用的助手) - if no agents
- 10. Skills Usage (可用资源/3. 技能 + 使用要求)
- 11. Constraint (资源使用要求)
- 12. Code Norms (python代码规范)
- 13. Footer (示例模板 + 结尾)
+ 6. Available Resources Header (可用资源 heading)
+ 7. Tools (可用资源/1. 工具 + 文件链接指南)
+ 8. Knowledge Base (知识库) - if knowledge_base_summary exists
+ 9. Managed Agents (可用资源/2. 助手) - if managed_agents exist
+ 10. External Agents (外部助手) - if external_a2a_agents exist
+ 11. Agent Fallback (当前没有可用的助手) - if no agents
+ 12. Skills Usage (可用资源/3. 技能 + 使用要求)
+ 13. Constraint (资源使用要求)
+ 14. Code Norms (python代码规范)
+ 15. Footer (示例模板 + 结尾)
Note: The a330d815 short-circuit (if system_prompt: return [single])
has been REMOVED. All callers must provide raw params for piecewise assembly.
@@ -1215,6 +1291,7 @@ def build_context_components(
build_skeleton_duty_component(
duty=duty,
language=language,
+ is_manager=is_manager,
)
)
@@ -1236,7 +1313,15 @@ def build_context_components(
)
)
- # 6. Tools + File URL Guide
+ # 6. Available Resources Header
+ components.append(
+ build_available_resources_header_component(
+ is_manager=is_manager,
+ language=language,
+ )
+ )
+
+ # 7. Tools + File URL Guide
if include_tools and tools:
components.append(
build_tools_component(
@@ -1247,7 +1332,17 @@ def build_context_components(
)
)
- # 7. Managed Agents (if exists) - manager only
+ # 8. Knowledge Base (if exists)
+ if include_knowledge_base and knowledge_base_summary:
+ components.append(
+ build_knowledge_base_component(
+ knowledge_base_summary=knowledge_base_summary,
+ kb_ids=kb_ids,
+ language=language,
+ )
+ )
+
+ # 9. Managed Agents (if exists) - manager only
if is_manager and include_managed_agents and managed_agents:
components.append(
build_managed_agents_component(
@@ -1256,7 +1351,7 @@ def build_context_components(
)
)
- # 8. External Agents (if exists) - manager only
+ # 10. External Agents (if exists) - manager only
if is_manager and include_external_agents and external_a2a_agents:
components.append(
build_external_agents_component(
@@ -1265,7 +1360,7 @@ def build_context_components(
)
)
- # 9. Agent Fallback (if no agents available) - manager only
+ # 11. Agent Fallback (if no agents available) - manager only
if is_manager and not managed_agents and not external_a2a_agents:
fallback_comp = build_agent_fallback_component(
managed_agents=managed_agents or {},
@@ -1275,16 +1370,17 @@ def build_context_components(
if fallback_comp.content: # Only add if has content
components.append(fallback_comp)
- # 10. Skills Usage Requirements
+ # 12. Skills Usage Requirements
if include_skills:
components.append(
build_skills_usage_component(
skills=skills or [],
language=language,
+ is_manager=is_manager,
)
)
- # 11. Constraint
+ # 13. Constraint
if constraint:
components.append(
build_skeleton_constraint_component(
@@ -1293,7 +1389,7 @@ def build_context_components(
)
)
- # 12. Code Norms
+ # 14. Code Norms
components.append(
build_skeleton_code_norms_component(
language=language,
@@ -1301,7 +1397,7 @@ def build_context_components(
)
)
- # 13. Footer
+ # 15. Footer
if few_shots:
components.append(
build_skeleton_footer_component(
diff --git a/sdk/nexent/core/agents/agent_context.py b/sdk/nexent/core/agents/agent_context.py
index 0b40d325c..6cb683a45 100644
--- a/sdk/nexent/core/agents/agent_context.py
+++ b/sdk/nexent/core/agents/agent_context.py
@@ -1343,6 +1343,26 @@ def get_registered_components(self) -> List:
with self._lock:
return list(self._components)
+ def replace_components(self, components: List) -> None:
+ """Atomically replace all registered components.
+
+ Clears existing components and registers new ones under a single
+ lock acquisition, preventing race conditions when the ContextManager
+ is shared across concurrent runs (e.g., conversation-level CM reuse).
+
+ Args:
+ components: List of ContextComponent instances to register.
+ Pass empty list to clear all components.
+ """
+ with self._lock:
+ self._components.clear()
+ for component in components:
+ if component.token_estimate == 0:
+ component.token_estimate = component.estimate_tokens(
+ self.config.chars_per_token
+ )
+ self._components.append(component)
+
def _get_strategy(self):
"""Factory method to get strategy instance based on config."""
from .agent_model import (
diff --git a/sdk/nexent/core/agents/core_agent.py b/sdk/nexent/core/agents/core_agent.py
index 73b13dacc..a6f3c0005 100644
--- a/sdk/nexent/core/agents/core_agent.py
+++ b/sdk/nexent/core/agents/core_agent.py
@@ -486,7 +486,12 @@ def run(self, task: str, stream: bool = False, reset: bool = True, images: Optio
{str(additional_args)}."""
system_prompt_content = self.system_prompt
- if self.context_manager and self.context_manager.get_registered_components():
+ registered = self.context_manager.get_registered_components() if self.context_manager else []
+ if registered:
+ self.logger.log(
+ f"ContextManager component path active: "
+ f"{[f'{c.component_type}(priority={c.priority},tokens={c.token_estimate})' for c in registered]}"
+ )
component_messages = self.context_manager.build_system_prompt()
if component_messages:
system_prompt_content = "\n\n".join(
diff --git a/sdk/nexent/core/agents/nexent_agent.py b/sdk/nexent/core/agents/nexent_agent.py
index b3c5b8cd0..907efe5e6 100644
--- a/sdk/nexent/core/agents/nexent_agent.py
+++ b/sdk/nexent/core/agents/nexent_agent.py
@@ -429,9 +429,9 @@ def create_single_agent(self, agent_config: AgentConfig):
)
agent.stop_event = self.stop_event
- # Mount context manager if config provided
+ # Mount context manager if config provided and enabled
ctx_config = getattr(agent_config, 'context_manager_config', None)
- if ctx_config:
+ if ctx_config and ctx_config.enabled:
agent.context_manager = ContextManager(
config=ctx_config,
max_steps=agent_config.max_steps
diff --git a/sdk/nexent/core/agents/run_agent.py b/sdk/nexent/core/agents/run_agent.py
index 243ca099e..69facc5cd 100644
--- a/sdk/nexent/core/agents/run_agent.py
+++ b/sdk/nexent/core/agents/run_agent.py
@@ -88,6 +88,8 @@ def agent_run_thread(agent_run_info: AgentRunInfo):
if getattr(agent_run_info, 'context_manager', None) is not None:
agent.context_manager = agent_run_info.context_manager
+ context_components = getattr(agent_run_info.agent_config, 'context_components', None)
+ agent.context_manager.replace_components(context_components or [])
nexent.add_history_to_agent(agent_run_info.history)
nexent.agent_run_with_observer(
@@ -109,6 +111,8 @@ def agent_run_thread(agent_run_info: AgentRunInfo):
if getattr(agent_run_info, 'context_manager', None) is not None:
agent.context_manager = agent_run_info.context_manager
+ context_components = getattr(agent_run_info.agent_config, 'context_components', None)
+ agent.context_manager.replace_components(context_components or [])
nexent.add_history_to_agent(agent_run_info.history)
nexent.agent_run_with_observer(
diff --git a/sdk/nexent/core/agents/summary_config.py b/sdk/nexent/core/agents/summary_config.py
index e271ddd34..8a568af5d 100644
--- a/sdk/nexent/core/agents/summary_config.py
+++ b/sdk/nexent/core/agents/summary_config.py
@@ -103,7 +103,7 @@ class ContextManagerConfig:
component_budgets: Dict[str, int] = field(default_factory=lambda: {
"system_prompt": 4000,
"tools": 3000,
- "skills": 1000,
+ "skills": 4000,
"memory": 2000,
"knowledge_base": 1500,
"managed_agents": 500,
diff --git a/test/backend/agents/test_create_agent_info.py b/test/backend/agents/test_create_agent_info.py
index f650de5d7..cee475e0a 100644
--- a/test/backend/agents/test_create_agent_info.py
+++ b/test/backend/agents/test_create_agent_info.py
@@ -389,8 +389,8 @@ def test_get_skills_for_template_exception_handling(self):
)
assert result == []
- mock_logger.warning.assert_called_once()
- assert "Failed to get skills for template: Service unavailable" in mock_logger.warning.call_args[0][0]
+ mock_logger.error.assert_called_once()
+ assert "Failed to get skills for agent" in mock_logger.error.call_args[0][0]
def test_get_skills_for_template_with_version_no(self):
"""Test case with specific version number"""
@@ -2624,7 +2624,7 @@ async def test_create_agent_config_knowledge_base_summary_error(self):
await create_agent_config("agent_1", "tenant_1", "user_1", "zh", "test query")
# Verify that error was logged
- mock_logger.error.assert_called_with("Failed to build knowledge base summary: Test Error")
+ mock_logger.error.assert_any_call("Failed to build knowledge base summary: Test Error")
class TestCreateModelConfigList:
diff --git a/test/backend/database/test_agent_db.py b/test/backend/database/test_agent_db.py
index 336cb031e..b834180cf 100644
--- a/test/backend/database/test_agent_db.py
+++ b/test/backend/database/test_agent_db.py
@@ -123,7 +123,7 @@ def __init__(self):
self.prompt_template_name = None
self.group_ids = None
self.is_new = True
- self.enable_context_manager = False
+ self.enable_context_manager = True
self.greeting_message = None
self.example_questions = None
self.current_version_no = None
diff --git a/test/backend/utils/test_context_component_types.py b/test/backend/utils/test_context_component_types.py
new file mode 100644
index 000000000..0ab2b12d6
--- /dev/null
+++ b/test/backend/utils/test_context_component_types.py
@@ -0,0 +1,659 @@
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+TEST_ROOT = Path(__file__).resolve().parents[2]
+PROJECT_ROOT = TEST_ROOT.parent
+
+for _path in (str(PROJECT_ROOT), str(TEST_ROOT)):
+ if _path not in sys.path:
+ sys.path.insert(0, _path)
+
+_sdk_dir = str(PROJECT_ROOT / "sdk")
+if _sdk_dir not in sys.path:
+ sys.path.insert(0, _sdk_dir)
+
+_mem0_stubs = {
+ "mem0": MagicMock(),
+ "mem0.memory": MagicMock(),
+ "mem0.memory.main": MagicMock(),
+ "mem0.embeddings": MagicMock(),
+ "mem0.embeddings.base": MagicMock(),
+ "mem0.configs": MagicMock(),
+ "mem0.configs.embeddings": MagicMock(),
+ "mem0.configs.embeddings.base": MagicMock(),
+ "smolagents": MagicMock(),
+ "smolagents.memory": MagicMock(),
+ "smolagents.agents": MagicMock(),
+ "smolagents.tools": MagicMock(),
+ "smolagents.models": MagicMock(),
+ "smolagents.local_python_executor": MagicMock(),
+ "smolagents.utils": MagicMock(),
+ "smolagents.monitoring": MagicMock(),
+ "openai": MagicMock(),
+ "openai.types": MagicMock(),
+ "openai.types.chat": MagicMock(),
+ "openai.types.chat.chat_completion_message": MagicMock(),
+ "openai.types.chat.chat_completion": MagicMock(),
+ "openai.types.chat.completion_create_params": MagicMock(),
+ "tiktoken": MagicMock(),
+ "tiktoken.encoding_for_model": MagicMock(),
+ "websockets": MagicMock(),
+ "websockets.client": MagicMock(),
+ "websockets.server": MagicMock(),
+ "dashscope": MagicMock(),
+ "dashscope.audio": MagicMock(),
+ "dashscope.audio.asr": MagicMock(),
+ "requests": MagicMock(),
+ "requests.exceptions": MagicMock(),
+ "boto3": MagicMock(),
+ "boto3.exceptions": MagicMock(),
+ "botocore": MagicMock(),
+ "botocore.exceptions": MagicMock(),
+ "botocore.client": MagicMock(),
+ "minio": MagicMock(),
+ "minio.error": MagicMock(),
+ "docker": MagicMock(),
+ "docker.errors": MagicMock(),
+ "docker.types": MagicMock(),
+ "fastmcp": MagicMock(),
+ "fastmcp.client": MagicMock(),
+ "fastmcp.client.transports": MagicMock(),
+ "kubernetes": MagicMock(),
+ "kubernetes.client": MagicMock(),
+ "kubernetes.config": MagicMock(),
+ "rich": MagicMock(),
+ "rich.console": MagicMock(),
+ "rich.markdown": MagicMock(),
+ "rich.panel": MagicMock(),
+ "rich.text": MagicMock(),
+}
+for _mod, _mock in _mem0_stubs.items():
+ if _mod not in sys.modules:
+ sys.modules[_mod] = _mock
+
+_nexent_sub_stubs = {
+ "nexent.memory": MagicMock(),
+ "nexent.memory.memory_core": MagicMock(),
+ "nexent.memory.memory_service": MagicMock(),
+ "nexent.memory.embedder_adaptor": MagicMock(),
+ "nexent.datamate": MagicMock(),
+ "nexent.datamate.datamate_client": MagicMock(),
+ "nexent.storage": MagicMock(),
+ "nexent.storage.storage_client_factory": MagicMock(),
+ "nexent.storage.minio": MagicMock(),
+ "nexent.storage.local": MagicMock(),
+ "nexent.container": MagicMock(),
+ "nexent.container.container_client_factory": MagicMock(),
+ "nexent.container.docker_client": MagicMock(),
+ "nexent.container.k8s_client": MagicMock(),
+ "nexent.core.models": MagicMock(),
+ "nexent.core.models.openai_llm": MagicMock(),
+ "nexent.core.models.openai_long_context_model": MagicMock(),
+ "nexent.core.models.embedding_model": MagicMock(),
+ "nexent.core.models.ali_stt_model": MagicMock(),
+ "nexent.core.agents.core_agent": MagicMock(),
+ "nexent.core.agents.agent_context": MagicMock(),
+ "nexent.core.agents.summary_cache": MagicMock(),
+ "nexent.core.agents.summary_config": MagicMock(),
+ "nexent.skills": MagicMock(),
+ "nexent.skills.skill_loader": MagicMock(),
+}
+for _mod, _mock in _nexent_sub_stubs.items():
+ if _mod not in sys.modules:
+ sys.modules[_mod] = _mock
+
+import pytest
+
+
+class TestBuilderReturnTypes:
+ def test_build_skeleton_header_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_header_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_header_component(
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ )
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_skeleton_duty_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_duty_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_duty_component(duty="Help users.")
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_skeleton_execution_flow_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_execution_flow_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_execution_flow_component()
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_skeleton_constraint_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_constraint_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_constraint_component(constraint="Be helpful.")
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_skeleton_code_norms_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_code_norms_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_code_norms_component()
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_skeleton_footer_returns_system_prompt(self):
+ from backend.utils.context_utils import build_skeleton_footer_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_skeleton_footer_component(few_shots="Q: hi? A: Hello!")
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_tools_returns_tools_component(self):
+ from backend.utils.context_utils import build_tools_component
+ from nexent.core.agents.agent_model import ToolsComponent
+
+ comp = build_tools_component(tools={})
+ assert isinstance(comp, ToolsComponent)
+ assert comp.component_type == "tools"
+
+ def test_build_skills_returns_skills_component(self):
+ from backend.utils.context_utils import build_skills_component
+ from nexent.core.agents.agent_model import SkillsComponent
+
+ comp = build_skills_component(
+ skills=[{"name": "test", "description": "desc"}]
+ )
+ assert isinstance(comp, SkillsComponent)
+ assert comp.component_type == "skills"
+
+ def test_build_memory_returns_memory_component(self):
+ from backend.utils.context_utils import build_memory_component
+ from nexent.core.agents.agent_model import MemoryComponent
+
+ comp = build_memory_component(
+ memory_list=[{"memory": "test", "score": 0.9, "memory_level": "user"}]
+ )
+ assert isinstance(comp, MemoryComponent)
+ assert comp.component_type == "memory"
+
+ def test_build_knowledge_base_returns_kb_component(self):
+ from backend.utils.context_utils import build_knowledge_base_component
+ from nexent.core.agents.agent_model import KnowledgeBaseComponent
+
+ comp = build_knowledge_base_component(
+ knowledge_base_summary="KB text", kb_ids=["kb-1"]
+ )
+ assert isinstance(comp, KnowledgeBaseComponent)
+ assert comp.component_type == "knowledge_base"
+
+ def test_build_managed_agents_returns_managed_component(self):
+ from backend.utils.context_utils import build_managed_agents_component
+ from nexent.core.agents.agent_model import ManagedAgentsComponent
+
+ comp = build_managed_agents_component(managed_agents={})
+ assert isinstance(comp, ManagedAgentsComponent)
+ assert comp.component_type == "managed_agents"
+
+ def test_build_external_agents_returns_external_component(self):
+ from backend.utils.context_utils import build_external_agents_component
+ from nexent.core.agents.agent_model import ExternalAgentsComponent
+
+ comp = build_external_agents_component(external_a2a_agents={})
+ assert isinstance(comp, ExternalAgentsComponent)
+ assert comp.component_type == "external_a2a_agents"
+
+ def test_build_skills_usage_returns_skills_component(self):
+ from backend.utils.context_utils import build_skills_usage_component
+ from nexent.core.agents.agent_model import SkillsComponent
+
+ comp = build_skills_usage_component(
+ skills=[{"name": "test", "description": "desc"}]
+ )
+ assert isinstance(comp, SkillsComponent)
+ assert comp.component_type == "skills"
+
+ def test_build_agent_fallback_returns_system_prompt(self):
+ from backend.utils.context_utils import build_agent_fallback_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_agent_fallback_component(
+ managed_agents={}, external_a2a_agents={}
+ )
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_build_available_resources_header_returns_system_prompt(self):
+ from backend.utils.context_utils import build_available_resources_header_component
+ from nexent.core.agents.agent_model import SystemPromptComponent
+
+ comp = build_available_resources_header_component()
+ assert isinstance(comp, SystemPromptComponent)
+ assert comp.component_type == "system_prompt"
+
+ def test_execution_flow_managed_text(self):
+ from backend.utils.context_utils import build_skeleton_execution_flow_component
+
+ comp = build_skeleton_execution_flow_component(is_manager=False, language="zh")
+ assert "确定需要使用哪些工具" in comp.content
+ assert "注意最后生成的回答要语义连贯" in comp.content
+
+ def test_execution_flow_manager_text(self):
+ from backend.utils.context_utils import build_skeleton_execution_flow_component
+
+ comp = build_skeleton_execution_flow_component(is_manager=True, language="zh")
+ assert "分析当前任务状态和进展" in comp.content
+ assert "分配给助手" in comp.content
+
+ def test_duty_managed_zh(self):
+ from backend.utils.context_utils import build_skeleton_duty_component
+
+ comp = build_skeleton_duty_component(duty="test", is_manager=False, language="zh")
+ assert "严禁直接执行代码" in comp.content
+
+ def test_duty_manager_zh(self):
+ from backend.utils.context_utils import build_skeleton_duty_component
+
+ comp = build_skeleton_duty_component(duty="test", is_manager=True, language="zh")
+ assert "文件操作必须使用平台提供的专用工具" in comp.content
+
+ def test_kb_not_duplicated_in_tools(self):
+ from backend.utils.context_utils import build_tools_component
+
+ class MockTool:
+ name = "t"
+ description = "Test tool"
+ inputs = "{}"
+ output_type = "str"
+ source = "local"
+
+ comp = build_tools_component(
+ tools={"t": MockTool()},
+ knowledge_base_summary="KB text",
+ )
+ assert "KB text" not in comp.formatted_description
+
+ def test_available_resources_header_manager(self):
+ from backend.utils.context_utils import build_available_resources_header_component
+
+ comp = build_available_resources_header_component(is_manager=True, language="zh")
+ assert "你只能使用以下资源" in comp.content
+
+ def test_available_resources_header_managed(self):
+ from backend.utils.context_utils import build_available_resources_header_component
+
+ comp = build_available_resources_header_component(is_manager=False, language="zh")
+ assert comp.content == "### 可用资源"
+
+
+class TestBuildContextComponentsAssembly:
+ def test_knowledge_base_included_when_flag_true_and_summary_exists(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ include_knowledge_base=True,
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ types = [c.component_type for c in components]
+ assert "knowledge_base" in types
+
+ def test_knowledge_base_excluded_when_flag_false(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ include_knowledge_base=False,
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ types = [c.component_type for c in components]
+ assert "knowledge_base" not in types
+
+ def test_knowledge_base_excluded_when_summary_empty(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ include_knowledge_base=True,
+ knowledge_base_summary="",
+ kb_ids=["kb-1"],
+ )
+ types = [c.component_type for c in components]
+ assert "knowledge_base" not in types
+
+ def test_skills_usage_has_skills_type(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ skills=[{"name": "s1", "description": "d1"}],
+ )
+ skills_components = [c for c in components if c.component_type == "skills"]
+ assert len(skills_components) >= 1
+ skills_usage = [
+ c
+ for c in skills_components
+ if hasattr(c, "skills") and c.skills == [{"name": "s1", "description": "d1"}]
+ ]
+ assert len(skills_usage) >= 1
+ assert skills_usage[0].component_type == "skills"
+
+ def test_all_component_types_present_with_full_inputs(self):
+ from backend.utils.context_utils import build_context_components
+
+ class MockTool:
+ name = "tool1"
+ description = "Test tool"
+ inputs = "{}"
+ output_type = "str"
+ source = "local"
+
+ class MockManagedAgent:
+ name = "agent1"
+ description = "Test agent"
+
+ class MockExternalAgent:
+ agent_id = "ext-1"
+ name = "External"
+ description = "External agent"
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ tools={"tool1": MockTool()},
+ skills=[{"name": "s1", "description": "d1"}],
+ managed_agents={"agent1": MockManagedAgent()},
+ external_a2a_agents={"ext-1": MockExternalAgent()},
+ memory_list=[{"memory": "test", "score": 0.9, "memory_level": "user"}],
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ types = [c.component_type for c in components]
+ assert "system_prompt" in types
+ assert "memory" in types
+ assert "skills" in types
+ assert "tools" in types
+ assert "managed_agents" in types
+ assert "external_a2a_agents" in types
+
+ def test_component_order_preserved(self):
+ from backend.utils.context_utils import build_context_components
+
+ class MockTool:
+ name = "tool1"
+ description = "Test tool"
+ inputs = "{}"
+ output_type = "str"
+ source = "local"
+
+ class MockManagedAgent:
+ name = "agent1"
+ description = "Test agent"
+
+ class MockExternalAgent:
+ agent_id = "ext-1"
+ name = "External"
+ description = "External agent"
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ tools={"tool1": MockTool()},
+ skills=[{"name": "s1", "description": "d1"}],
+ managed_agents={"agent1": MockManagedAgent()},
+ external_a2a_agents={"ext-1": MockExternalAgent()},
+ memory_list=[{"memory": "test", "score": 0.9, "memory_level": "user"}],
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ types = [c.component_type for c in components]
+ expected_order = [
+ "system_prompt",
+ "memory",
+ "system_prompt",
+ "skills",
+ "system_prompt",
+ "system_prompt",
+ "tools",
+ "knowledge_base",
+ "managed_agents",
+ "external_a2a_agents",
+ "skills",
+ "system_prompt",
+ "system_prompt",
+ "system_prompt",
+ ]
+ assert types == expected_order
+
+ def test_kb_ids_passed_through(self):
+ from backend.utils.context_utils import build_context_components
+ from nexent.core.agents.agent_model import KnowledgeBaseComponent
+
+ components = build_context_components(
+ duty="Help users.",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ kb_ids=["kb-1", "kb-2"],
+ knowledge_base_summary="text",
+ )
+ kb_components = [
+ c for c in components if isinstance(c, KnowledgeBaseComponent)
+ ]
+ assert len(kb_components) >= 1
+ assert kb_components[0].kb_ids == ["kb-1", "kb-2"]
+
+
+class TestComponentToMessages:
+ def test_skills_component_to_messages(self):
+ from nexent.core.agents.agent_model import SkillsComponent
+
+ comp = SkillsComponent(
+ skills=[{"name": "test", "description": "desc"}],
+ formatted_description="test desc",
+ )
+ messages = comp.to_messages()
+ assert messages == [{"role": "system", "content": "test desc"}]
+
+ def test_knowledge_base_component_to_messages(self):
+ from nexent.core.agents.agent_model import KnowledgeBaseComponent
+
+ comp = KnowledgeBaseComponent(summary="KB summary")
+ messages = comp.to_messages()
+ assert messages == [{"role": "system", "content": "KB summary"}]
+
+ def test_knowledge_base_component_empty_summary_no_messages(self):
+ from nexent.core.agents.agent_model import KnowledgeBaseComponent
+
+ comp = KnowledgeBaseComponent(summary="")
+ messages = comp.to_messages()
+ assert messages == []
+
+ def test_memory_component_to_messages(self):
+ from nexent.core.agents.agent_model import MemoryComponent
+
+ comp = MemoryComponent(formatted_content="memory text")
+ messages = comp.to_messages()
+ assert messages == [{"role": "system", "content": "memory text"}]
+
+ def test_tools_component_to_messages(self):
+ from nexent.core.agents.agent_model import ToolsComponent
+
+ comp = ToolsComponent(formatted_description="tools text")
+ messages = comp.to_messages()
+ assert messages == [{"role": "system", "content": "tools text"}]
+
+
+class TestFullPromptAssembly:
+ def test_full_assembly_produces_system_messages(self):
+ from backend.utils.context_utils import build_context_components
+
+ class MockTool:
+ name = "tool1"
+ description = "Test tool"
+ inputs = "{}"
+ output_type = "str"
+ source = "local"
+
+ class MockManagedAgent:
+ name = "agent1"
+ description = "Test agent"
+
+ class MockExternalAgent:
+ agent_id = "ext-1"
+ name = "External"
+ description = "External agent"
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ tools={"tool1": MockTool()},
+ skills=[{"name": "s1", "description": "d1"}],
+ managed_agents={"agent1": MockManagedAgent()},
+ external_a2a_agents={"ext-1": MockExternalAgent()},
+ memory_list=[{"memory": "test", "score": 0.9, "memory_level": "user"}],
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ all_messages = []
+ for comp in components:
+ all_messages.extend(comp.to_messages())
+ assert len(all_messages) > 0
+ for msg in all_messages:
+ assert msg["role"] == "system"
+ assert msg["content"]
+
+ def test_full_assembly_contains_key_sections(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ )
+ all_messages = []
+ for comp in components:
+ all_messages.extend(comp.to_messages())
+ combined = "\n".join(msg["content"] for msg in all_messages)
+ assert "\u57fa\u672c\u4fe1\u606f" in combined or "Basic Information" in combined
+ assert "\u6838\u5fc3\u804c\u8d23" in combined or "Core Responsibilities" in combined
+ assert "\u6267\u884c\u6d41\u7a0b" in combined or "Execution Process" in combined
+ assert "python\u4ee3\u7801\u89c4\u8303" in combined or "Python Code Specifications" in combined
+ assert "\u53ef\u7528\u8d44\u6e90" in combined or "Available Resources" in combined
+
+ def test_english_language_produces_english_content(self):
+ from backend.utils.context_utils import build_context_components
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ language="en",
+ )
+ all_messages = []
+ for comp in components:
+ all_messages.extend(comp.to_messages())
+ combined = "\n".join(msg["content"] for msg in all_messages)
+ assert "Basic Information" in combined
+ assert "Core Responsibilities" in combined
+ assert "Execution Process" in combined
+
+ def test_component_count_matches_expected(self):
+ from backend.utils.context_utils import build_context_components
+
+ class MockTool:
+ name = "tool1"
+ description = "Test tool"
+ inputs = "{}"
+ output_type = "str"
+ source = "local"
+
+ class MockManagedAgent:
+ name = "agent1"
+ description = "Test agent"
+
+ class MockExternalAgent:
+ agent_id = "ext-1"
+ name = "External"
+ description = "External agent"
+
+ components = build_context_components(
+ duty="Help users.",
+ constraint="Be helpful.",
+ few_shots="Q: hi? A: Hello!",
+ app_name="Test",
+ app_description="Desc",
+ time_str="2026-01-01",
+ user_id="u1",
+ is_manager=True,
+ tools={"tool1": MockTool()},
+ skills=[{"name": "s1", "description": "d1"}],
+ managed_agents={"agent1": MockManagedAgent()},
+ external_a2a_agents={"ext-1": MockExternalAgent()},
+ memory_list=[{"memory": "test", "score": 0.9, "memory_level": "user"}],
+ knowledge_base_summary="KB text",
+ kb_ids=["kb-1"],
+ )
+ assert len(components) == 14
+
+
+if __name__ == "__main__":
+ pytest.main([__file__])
diff --git a/test/backend/utils/test_context_utils.py b/test/backend/utils/test_context_utils.py
index 66e789477..0debf20de 100644
--- a/test/backend/utils/test_context_utils.py
+++ b/test/backend/utils/test_context_utils.py
@@ -14,7 +14,12 @@ class TestFormatFunctions:
def test_format_tools_empty(self):
from backend.utils.context_utils import _format_tools_description
result = _format_tools_description({}, language="zh")
- assert result == "- 当前没有可用的工具"
+ assert result == "1. 工具\n- 当前没有可用的工具"
+
+ def test_format_tools_empty_managed(self):
+ from backend.utils.context_utils import _format_tools_description
+ result = _format_tools_description({}, language="zh", is_manager=False)
+ assert result == "1. 工具\n- 当前没有可用的工具"
def test_format_tools_single(self):
from backend.utils.context_utils import _format_tools_description
@@ -130,7 +135,8 @@ def test_build_knowledge_base_component_empty(self):
def test_build_knowledge_base_component_with_summary(self):
from backend.utils.context_utils import build_knowledge_base_component
comp = build_knowledge_base_component("KB text", kb_ids=["kb-1"])
- assert comp.summary == "KB text"
+ assert "KB text" in comp.summary
+ assert "knowledge_base_search" in comp.summary
def test_build_managed_agents_component_empty(self):
from backend.utils.context_utils import build_managed_agents_component
diff --git a/test/sdk/core/agents/test_agent_context/unit/test_component_management.py b/test/sdk/core/agents/test_agent_context/unit/test_component_management.py
index 5f25e1119..8e4304044 100644
--- a/test/sdk/core/agents/test_agent_context/unit/test_component_management.py
+++ b/test/sdk/core/agents/test_agent_context/unit/test_component_management.py
@@ -98,6 +98,53 @@ def test_clear_allows_new_registration(self):
assert cm.get_registered_components()[0]._content == "new"
+class TestReplaceComponents:
+ """Tests for replace_components() atomic swap method."""
+
+ def test_replace_on_empty_manager(self):
+ cm = ContextManager()
+ cm.replace_components([MockComponent(content="new1"), MockComponent(content="new2")])
+ assert len(cm.get_registered_components()) == 2
+
+ def test_replace_clears_existing(self):
+ cm = ContextManager()
+ cm.register_component(MockComponent(content="old1"))
+ cm.register_component(MockComponent(content="old2"))
+ cm.replace_components([MockComponent(content="new")])
+ registered = cm.get_registered_components()
+ assert len(registered) == 1
+ assert registered[0]._content == "new"
+
+ def test_replace_with_empty_list(self):
+ cm = ContextManager()
+ cm.register_component(MockComponent(content="old"))
+ cm.replace_components([])
+ assert cm.get_registered_components() == []
+
+ def test_replace_estimates_tokens(self):
+ cm = ContextManager()
+ comp = MockComponent(content="some content here", token_estimate=0)
+ cm.replace_components([comp])
+ assert cm.get_registered_components()[0].token_estimate > 0
+
+ def test_replace_preserves_existing_token_estimate(self):
+ cm = ContextManager()
+ comp = MockComponent(content="x", token_estimate=42)
+ cm.replace_components([comp])
+ assert cm.get_registered_components()[0].token_estimate == 42
+
+ def test_replace_preserves_order(self):
+ cm = ContextManager()
+ comps = [
+ MockComponent(content="first", priority=10),
+ MockComponent(content="second", priority=20),
+ MockComponent(content="third", priority=30),
+ ]
+ cm.replace_components(comps)
+ registered = cm.get_registered_components()
+ assert [c._content for c in registered] == ["first", "second", "third"]
+
+
class TestGetRegisteredComponents:
"""Tests for get_registered_components() method."""
diff --git a/test/sdk/core/agents/test_nexent_agent_component_integration.py b/test/sdk/core/agents/test_nexent_agent_component_integration.py
index 49483d94b..acd31f584 100644
--- a/test/sdk/core/agents/test_nexent_agent_component_integration.py
+++ b/test/sdk/core/agents/test_nexent_agent_component_integration.py
@@ -29,6 +29,7 @@ def mock_context_manager(self):
@pytest.fixture
def agent_config_with_components(self):
ctx_config = ContextManagerConfig(
+ enabled=True,
token_threshold=1000,
strategy=STRATEGY_TOKEN_BUDGET,
component_budgets={"tools": 200, "skills": 100},
@@ -53,7 +54,7 @@ def test_context_manager_mounted_when_config_present(self, agent_config_with_com
agent.context_manager = None
ctx_config = getattr(agent_config_with_components, 'context_manager_config', None)
- if ctx_config:
+ if ctx_config and ctx_config.enabled:
from sdk.nexent.core.agents.agent_context import ContextManager
agent.context_manager = ContextManager(
config=ctx_config,
@@ -83,6 +84,26 @@ def test_no_context_manager_when_config_absent(self):
assert ctx_config is None
assert agent.context_manager is None
+ def test_no_context_manager_when_config_disabled(self):
+ ctx_config = ContextManagerConfig(enabled=False, token_threshold=1000)
+ agent_config = AgentConfig(
+ name="test_agent",
+ description="Test agent",
+ model_name="test-model",
+ tools=[],
+ context_manager_config=ctx_config,
+ )
+
+ agent = MagicMock()
+ agent.context_manager = None
+
+ config = getattr(agent_config, 'context_manager_config', None)
+ if config and config.enabled:
+ from sdk.nexent.core.agents.agent_context import ContextManager
+ agent.context_manager = ContextManager(config=config, max_steps=10)
+
+ assert agent.context_manager is None
+
def test_components_registered_in_order(self, mock_context_manager, agent_config_with_components):
components = getattr(agent_config_with_components, 'context_components', [])
@@ -196,4 +217,64 @@ def test_context_manager_config_without_strategy_defaults(self):
config = ContextManagerConfig(token_threshold=2000)
assert config.strategy == STRATEGY_TOKEN_BUDGET
- assert "system_prompt" in config.component_budgets
\ No newline at end of file
+ assert "system_prompt" in config.component_budgets
+
+
+class TestConversationLevelCMComponentSurvival:
+ """Tests verifying components survive conversation-level CM overwrite."""
+
+ def test_replace_components_after_overwrite(self):
+ from sdk.nexent.core.agents.agent_context import ContextManager
+
+ conversation_cm = ContextManager(
+ config=ContextManagerConfig(enabled=True, token_threshold=1000),
+ max_steps=10,
+ )
+ assert conversation_cm.get_registered_components() == []
+
+ components = [
+ ToolsComponent(content="Tool descriptions", token_estimate=50),
+ SystemPromptComponent(content="System prompt", token_estimate=100),
+ ]
+
+ conversation_cm.replace_components(components)
+
+ registered = conversation_cm.get_registered_components()
+ assert len(registered) == 2
+ assert registered[0].component_type == "tools"
+ assert registered[1].component_type == "system_prompt"
+
+ def test_replace_components_clears_stale(self):
+ from sdk.nexent.core.agents.agent_context import ContextManager
+
+ conversation_cm = ContextManager(
+ config=ContextManagerConfig(enabled=True, token_threshold=1000),
+ max_steps=10,
+ )
+ conversation_cm.register_component(
+ ToolsComponent(content="stale tools", token_estimate=50)
+ )
+ assert len(conversation_cm.get_registered_components()) == 1
+
+ new_components = [
+ SystemPromptComponent(content="fresh prompt", token_estimate=100),
+ ]
+ conversation_cm.replace_components(new_components)
+
+ registered = conversation_cm.get_registered_components()
+ assert len(registered) == 1
+ assert registered[0].component_type == "system_prompt"
+
+ def test_replace_components_with_empty_list(self):
+ from sdk.nexent.core.agents.agent_context import ContextManager
+
+ conversation_cm = ContextManager(
+ config=ContextManagerConfig(enabled=True, token_threshold=1000),
+ max_steps=10,
+ )
+ conversation_cm.register_component(
+ ToolsComponent(content="tools", token_estimate=50)
+ )
+
+ conversation_cm.replace_components([])
+ assert conversation_cm.get_registered_components() == []
\ No newline at end of file