diff --git a/backend/adapters/__init__.py b/backend/adapters/__init__.py index ed46fc888e..6db039c552 100644 --- a/backend/adapters/__init__.py +++ b/backend/adapters/__init__.py @@ -1,9 +1,15 @@ from adapters.exception import JiuwenSDKError, JiuwenSDKUnavailableError, NexentCapabilityError -try: - from adapters.jiuwen_sdk_adapter import JiuwenSDKAdapter -except ModuleNotFoundError: - JiuwenSDKAdapter = None # type: ignore[assignment, misc] + +def __getattr__(name: str): + """Load the optional OpenJiuwen evaluation adapter only when requested.""" + if name != "JiuwenSDKAdapter": + raise AttributeError(name) + try: + from adapters.jiuwen_sdk_adapter import JiuwenSDKAdapter + except ModuleNotFoundError: + return None + return JiuwenSDKAdapter __all__ = [ "JiuwenSDKError", diff --git a/backend/agents/create_agent_info.py b/backend/agents/create_agent_info.py index 8a956cdc69..df0e53cc3d 100644 --- a/backend/agents/create_agent_info.py +++ b/backend/agents/create_agent_info.py @@ -6,7 +6,15 @@ from jinja2 import Template, StrictUndefined from nexent.core.utils.observer import MessageObserver -from nexent.core.agents.agent_model import AgentRunInfo, ModelConfig, AgentConfig, ToolConfig, ExternalA2AAgentConfig, AgentHistory, AgentVerificationConfig +from nexent.core.agents.agent_model import ( + AgentConfig, + AgentHistory, + AgentRunInfo, + AgentVerificationConfig, + ExternalA2AAgentConfig, + ModelConfig, + ToolConfig, +) from nexent.core.agents.summary_config import ContextManagerConfig from nexent.core.models.prompt_cache import resolve_prompt_cache_profile from nexent.core.models.capacity_resolver import ( @@ -53,7 +61,7 @@ from utils.context_utils import build_context_components from utils.redis_utils import get_redis_client from consts.const import LOCAL_MCP_SERVER, MODEL_CONFIG_MAPPING, LANGUAGE, DATA_PROCESS_SERVICE, MINIO_DEFAULT_BUCKET -from consts.model import AgentToolParamsRequest, ToolParamsRequest +from consts.model import ToolParamsRequest from consts.exceptions import ValidationError logger = logging.getLogger("create_agent_info") @@ -527,7 +535,8 @@ def _get_external_a2a_agents( def _get_skill_script_tools( agent_id: int, tenant_id: str, - version_no: int = 0 + version_no: int = 0, + allowed_skills: Optional[List[dict]] = None, ) -> List[ToolConfig]: """Get tool config for skill script execution and skill reading. @@ -535,6 +544,7 @@ def _get_skill_script_tools( agent_id: Agent ID for filtering available skills in error messages. tenant_id: Tenant ID for filtering available skills in error messages. version_no: Version number for filtering available skills. + allowed_skills: Skill summaries already authorized during agent assembly. Returns: List of ToolConfig for skill execution and reading tools @@ -546,6 +556,14 @@ def _get_skill_script_tools( "tenant_id": tenant_id, "version_no": version_no, } + if allowed_skills is not None: + skill_context["allowed_skills"] = sorted( + { + str(skill.get("name") or "").strip() + for skill in allowed_skills + if str(skill.get("name") or "").strip() + } + ) try: return [ @@ -706,10 +724,43 @@ async def create_agent_config( request_requested_output_tokens: int | None = None, tool_params: Optional[ToolParamsRequest | Dict[str, Any]] = None, enable_planning: bool = False, + _ancestry: tuple[int, ...] = (), + _expected_runtime_framework: str | None = None, ): normalized_tool_params = _normalize_tool_params_request(tool_params) agent_info = search_agent_info_by_agent_id( agent_id=agent_id, tenant_id=tenant_id, version_no=version_no) + if agent_id in _ancestry: + raise ValidationError( + f"Circular dependency detected while assembling Agent {agent_id}." + ) + if "runtime_framework" not in agent_info: + runtime_framework = "smolagents" + else: + runtime_framework = agent_info.get("runtime_framework") + if runtime_framework is None: + from consts.error_code import ErrorCode + from consts.exceptions import AppException + + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED, + f"Agent {agent_id} must select a runtime framework before it can run.", + ) + if runtime_framework not in {"smolagents", "openjiuwen"}: + raise ValidationError( + f"Agent {agent_id} has unsupported runtime framework: {runtime_framework}." + ) + if ( + _expected_runtime_framework is not None + and runtime_framework != _expected_runtime_framework + ): + from consts.error_code import ErrorCode + from consts.exceptions import AppException + + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_MISMATCH, + "Internal parent and child Agents must use the same runtime framework.", + ) # create sub agent sub_agent_relations = query_sub_agent_relations( @@ -732,6 +783,8 @@ async def create_agent_config( version_no=sub_agent_version_no, override_model_id=None, tool_params=normalized_tool_params, + _ancestry=(*_ancestry, agent_id), + _expected_runtime_framework=runtime_framework, ) managed_agents.append(sub_agent_config) @@ -914,7 +967,12 @@ async def create_agent_config( skills = _get_skills_for_template(agent_id, tenant_id, version_no) is_manager = len(managed_agents) > 0 or len(external_a2a_agents) > 0 - builtin_tools = _get_skill_script_tools(agent_id, tenant_id, version_no) + builtin_tools = _get_skill_script_tools( + agent_id, + tenant_id, + version_no, + allowed_skills=skills, + ) available_tools = tool_list + builtin_tools _inject_plan_tools(available_tools, enable_planning) @@ -1052,6 +1110,8 @@ async def create_agent_config( agent_config.enable_planning, any(t.name in {"create_plan", "update_plan_step"} for t in agent_config.tools), ) + agent_config.id = agent_id + agent_config.runtime_framework = runtime_framework return agent_config @@ -1450,6 +1510,73 @@ def check_agent_tools(agent_config: AgentConfig): return list(used_mcp_urls) +def attach_mcp_bindings( + input_agent_config: AgentConfig, + mcp_info_dict: Dict[str, dict], +) -> None: + """Attach per-node MCP bindings without re-querying MCP configuration.""" + from nexent.core.agents.agent_model import MCPBinding + + bindings: list[MCPBinding] = [] + tools_by_server: Dict[str, list[ToolConfig]] = {} + for tool in input_agent_config.tools: + if tool.source == "mcp" and tool.usage: + tools_by_server.setdefault(tool.usage, []).append(tool) + + for server_name, tools in tools_by_server.items(): + record = mcp_info_dict.get(server_name) or {} + url = str(record.get("remote_mcp_server") or "").strip() + headers: Dict[str, str] = {} + auth_token = record.get("authorization_token") + if auth_token: + headers["Authorization"] = str(auth_token) + custom_headers = record.get("custom_headers") + if isinstance(custom_headers, dict): + headers.update({str(key): str(value) for key, value in custom_headers.items()}) + tool_names = sorted( + { + str(tool.class_name or tool.name) + for tool in tools + if tool.class_name or tool.name + } + ) + required_tool_names = sorted( + { + str(tool.class_name or tool.name) + for tool in tools + if (tool.class_name or tool.name) + and bool((tool.metadata or {}).get("mcp_required", True)) + } + ) + available = bool(record.get("status") and url) + if not record: + unavailable_reason = "server_not_configured" + elif not record.get("status"): + unavailable_reason = "server_disabled" + elif not url: + unavailable_reason = "server_url_missing" + else: + unavailable_reason = None + bindings.append( + MCPBinding( + server_id=str(record.get("mcp_id") or server_name), + server_name=server_name, + url=url, + transport="sse" if url.endswith("/sse") else "streamable-http", + headers=headers, + required=bool(required_tool_names), + tool_names=tool_names, + required_tool_names=required_tool_names, + available=available, + unavailable_reason=unavailable_reason, + ) + ) + input_agent_config.mcp_bindings = bindings + + for sub_agent_config in input_agent_config.managed_agents: + attach_mcp_bindings(sub_agent_config, mcp_info_dict) + + async def create_agent_run_info( agent_id, minio_files, @@ -1510,10 +1637,18 @@ async def create_agent_run_info( "status": True, "authorization_token": None }) - remote_mcp_dict = {record["remote_mcp_server_name"]: record for record in remote_mcp_list if record["status"]} + all_mcp_dict = {record["remote_mcp_server_name"]: record for record in remote_mcp_list} + enabled_mcp_dict = { + name: record + for name, record in all_mcp_dict.items() + if record.get("status") + } + is_agent_config = isinstance(AgentConfig, type) and isinstance(agent_config, AgentConfig) + if is_agent_config: + attach_mcp_bindings(agent_config, all_mcp_dict) # Filter MCP servers and tools, and build mcp_host with authorization - used_mcp_urls = filter_mcp_servers_and_tools(agent_config, remote_mcp_dict) + used_mcp_urls = filter_mcp_servers_and_tools(agent_config, enabled_mcp_dict) # Build mcp_host list with authorization tokens and custom headers mcp_host = [] @@ -1563,4 +1698,7 @@ async def create_agent_run_info( ), redis_client=get_redis_client(), ) + if is_agent_config: + agent_run_info.runtime_framework = agent_config.runtime_framework + agent_run_info.mcp_bindings = list(agent_config.mcp_bindings) return agent_run_info diff --git a/backend/apps/agent_app.py b/backend/apps/agent_app.py index a0aa9d3838..1c5a7a5327 100644 --- a/backend/apps/agent_app.py +++ b/backend/apps/agent_app.py @@ -9,7 +9,7 @@ from consts.const import ASSET_OWNER_TENANT_ID from consts.model import AgentRequest, AgentInfoRequest, AgentIDRequest, ConversationResponse, AgentImportRequest, AgentNameBatchCheckRequest, AgentNameBatchRegenerateRequest, VersionPublishRequest, VersionListResponse, VersionDetailResponse, VersionRollbackRequest, VersionStatusRequest, CurrentVersionResponse, VersionCompareRequest, VersionUpdateRequest -from consts.exceptions import SkillDuplicateError +from consts.exceptions import AppException, SkillDuplicateError from services.asset_owner_visibility import apply_agent_detail_prompt_visibility from services.agent_service import ( @@ -70,6 +70,8 @@ async def agent_run_api( authorization=authorization, resume=resume, ) + except AppException: + raise except Exception as e: logger.error(f"Agent run error: {str(e)}") # Only expose actual error in debug mode for better diagnosis @@ -155,6 +157,8 @@ async def update_agent_info_api(request: AgentInfoRequest, authorization: Option try: result = await update_agent_info_impl(request, authorization) return result or {} + except AppException: + raise except Exception as e: logger.error(f"Agent update error: {str(e)}") raise HTTPException( @@ -234,6 +238,8 @@ async def import_agent_api(request: AgentImportRequest, authorization: Optional[ force_import=request.force_import ) return {} + except AppException: + raise except SkillDuplicateError as exc: raise HTTPException(status_code=409, detail={ "type": "skill_duplicate", @@ -357,6 +363,8 @@ async def publish_version_api( publish_as_a2a=request.publish_as_a2a, ) return JSONResponse(status_code=HTTPStatus.OK, content=result) + except AppException: + raise except ValueError as e: raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) except Exception as e: @@ -489,6 +497,8 @@ async def rollback_version_api( target_version_no=version_no, ) return JSONResponse(status_code=HTTPStatus.OK, content=result) + except AppException: + raise except ValueError as e: raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) except Exception as e: @@ -627,5 +637,3 @@ async def list_published_agents_api( raise HTTPException( status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Published agents list error." ) - - diff --git a/backend/apps/runtime_app.py b/backend/apps/runtime_app.py index d623b14e09..ea2f77375e 100644 --- a/backend/apps/runtime_app.py +++ b/backend/apps/runtime_app.py @@ -31,6 +31,14 @@ app.include_router(skill_creator_router) +@app.on_event("shutdown") +async def shutdown_agent_runtimes() -> None: + """Drain initialized in-process runtimes without importing unused providers.""" + from services.agent_runtime.registry import shutdown_initialized_runtimes + + await shutdown_initialized_runtimes() + + @app.on_event("startup") async def start_agent_automation_scheduler(): from services.agent_automation.scheduler import agent_automation_scheduler diff --git a/backend/consts/agent_runtime.py b/backend/consts/agent_runtime.py new file mode 100644 index 0000000000..b5cadfcdde --- /dev/null +++ b/backend/consts/agent_runtime.py @@ -0,0 +1,37 @@ +"""Agent-level runtime framework constants and normalization helpers.""" + +from enum import Enum + + +class AgentRuntimeFramework(str, Enum): + """Execution framework persisted with every Agent version.""" + + SMOLAGENTS = "smolagents" + OPENJIUWEN = "openjiuwen" + + +DEFAULT_AGENT_RUNTIME_FRAMEWORK = AgentRuntimeFramework.SMOLAGENTS.value +SUPPORTED_AGENT_RUNTIME_FRAMEWORKS = frozenset(item.value for item in AgentRuntimeFramework) + + +def normalize_agent_runtime_framework( + value: str | AgentRuntimeFramework | None, + *, + default: str | None = DEFAULT_AGENT_RUNTIME_FRAMEWORK, +) -> str | None: + """Normalize a persisted or request value and reject unsupported frameworks.""" + if value is None: + return default + normalized = value.value if isinstance(value, AgentRuntimeFramework) else str(value).strip().lower() + if normalized not in SUPPORTED_AGENT_RUNTIME_FRAMEWORKS: + allowed = ", ".join(sorted(SUPPORTED_AGENT_RUNTIME_FRAMEWORKS)) + raise ValueError(f"Unsupported runtime_framework {value!r}; allowed values: {allowed}.") + return normalized + + +__all__ = [ + "AgentRuntimeFramework", + "DEFAULT_AGENT_RUNTIME_FRAMEWORK", + "SUPPORTED_AGENT_RUNTIME_FRAMEWORKS", + "normalize_agent_runtime_framework", +] diff --git a/backend/consts/error_code.py b/backend/consts/error_code.py index a6326668fd..a619bc144c 100644 --- a/backend/consts/error_code.py +++ b/backend/consts/error_code.py @@ -77,6 +77,9 @@ class ErrorCode(Enum): AGENTSPACE_AGENT_RUN_FAILED = "030103" # Agent run failed AGENTSPACE_AGENT_NAME_DUPLICATE = "030104" # Duplicate agent name AGENTSPACE_VERSION_NOT_FOUND = "030105" # Agent version not found + AGENT_RUNTIME_FRAMEWORK_IMMUTABLE = "030106" # Runtime framework cannot be changed after creation + AGENT_RUNTIME_FRAMEWORK_MISMATCH = "030107" # Internal parent and child frameworks differ + AGENT_RUNTIME_FRAMEWORK_REQUIRED = "030108" # Blank agent has no selected runtime framework # ==================== 04 AgentMarket / 智能体市场 ==================== # 01 - Agent @@ -245,6 +248,9 @@ class ErrorCode(Enum): ErrorCode.COMMON_RESOURCE_NOT_FOUND: 404, ErrorCode.COMMON_RESOURCE_ALREADY_EXISTS: 409, ErrorCode.COMMON_RESOURCE_DISABLED: 403, + ErrorCode.AGENT_RUNTIME_FRAMEWORK_IMMUTABLE: 409, + ErrorCode.AGENT_RUNTIME_FRAMEWORK_MISMATCH: 409, + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED: 409, # Common - File ErrorCode.FILE_NOT_FOUND: 404, ErrorCode.FILE_UPLOAD_FAILED: 500, diff --git a/backend/consts/model.py b/backend/consts/model.py index f2484371c1..cf12bc303d 100644 --- a/backend/consts/model.py +++ b/backend/consts/model.py @@ -573,6 +573,7 @@ class AgentInfoRequest(BaseModel): verification_config: Optional[Dict[str, Any]] = None greeting_message: Optional[str] = None example_questions: Optional[List[str]] = None + runtime_framework: Optional[Literal["smolagents", "openjiuwen"]] = None version_no: int = 0 @field_validator("verification_config", mode="before") @@ -671,6 +672,7 @@ class ExportAndImportAgentInfo(BaseModel): skill_names: Optional[List[str]] = None prompt_template_id: Optional[int] = None prompt_template_name: Optional[str] = None + runtime_framework: Literal["smolagents", "openjiuwen"] = "smolagents" class Config: arbitrary_types_allowed = True diff --git a/backend/database/agent_db.py b/backend/database/agent_db.py index e6c76f85b9..44e893328d 100644 --- a/backend/database/agent_db.py +++ b/backend/database/agent_db.py @@ -241,6 +241,7 @@ def create_agent(agent_info, tenant_id: str, user_id: str): "verification_config": new_agent.verification_config, "greeting_message": new_agent.greeting_message, "example_questions": new_agent.example_questions, + "runtime_framework": new_agent.runtime_framework, "current_version_no": new_agent.current_version_no, "version_no": new_agent.version_no, "created_by": new_agent.created_by, diff --git a/backend/database/db_models.py b/backend/database/db_models.py index 899084a895..aa1f477373 100644 --- a/backend/database/db_models.py +++ b/backend/database/db_models.py @@ -623,6 +623,11 @@ class AgentInfo(TableBase): verification_config = Column(JSONB, doc="Layered ReAct self-verification configuration") 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") + runtime_framework = Column( + String(20), + nullable=True, + doc="Immutable execution framework: smolagents or openjiuwen", + ) class PromptTemplate(TableBase): diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 78639250b4..955733fd1c 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -26,7 +26,7 @@ dependencies = [ "scikit-learn>=1.0.0", "numpy>=1.24.0", "defusedxml>=0.7.1", - "openjiuwen>=0.1.0", + "openjiuwen==0.1.16", "pydantic-settings>=2.0.0", "python-docx>=1.1.0", "xlrd>=2.0.1", diff --git a/backend/services/agent_evaluation_service.py b/backend/services/agent_evaluation_service.py index 49c1996234..5ec9eac4d8 100644 --- a/backend/services/agent_evaluation_service.py +++ b/backend/services/agent_evaluation_service.py @@ -2,15 +2,14 @@ import io import json import logging +import re from statistics import mean from typing import Any, Dict, List, Optional, Tuple -from adapters.exception import JiuwenSDKError, JiuwenSDKUnavailableError +from openpyxl import Workbook +from openpyxl.styles import Alignment, Font, PatternFill -try: - from adapters.jiuwen_sdk_adapter import JiuwenSDKAdapter -except ModuleNotFoundError: - JiuwenSDKAdapter = None # type: ignore[assignment, misc] +from adapters.exception import JiuwenSDKUnavailableError from consts.model import AgentRequest from database.agent_evaluation_db import ( create_agent_evaluation, @@ -26,13 +25,25 @@ from services.evaluation_set_service import resolve_latest_published_version_no from services.agent_service import prepare_agent_run from utils.thread_utils import pool -from openpyxl import Workbook -from openpyxl.styles import Font, PatternFill, Alignment -import re + +JiuwenSDKAdapter = None logger = logging.getLogger(__name__) +def _get_jiuwen_adapter_class(): + """Resolve the optional evaluation adapter without importing OpenJiuwen at startup.""" + global JiuwenSDKAdapter + if JiuwenSDKAdapter is not None: + return JiuwenSDKAdapter + try: + from adapters.jiuwen_sdk_adapter import JiuwenSDKAdapter as adapter_class + except ModuleNotFoundError: + return None + JiuwenSDKAdapter = adapter_class + return adapter_class + + # Log records emitted during SDK invocations may bleed into the ``reason`` # field as ``"[HH:MM:SS LEVEL logger_name] {...payload...}"``. Extract the # embedded JSON ``reason`` from those polluted strings so the report shows the @@ -47,6 +58,7 @@ r")\s*" ) _MARKDOWN_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL) +_JSON_OBJECT_RE = re.compile(r"\{.*\}", re.DOTALL) def _extract_clean_reason(raw: Any) -> str: @@ -572,10 +584,11 @@ def execute_agent_evaluation_run( raise ValueError("judge_model_id is required but neither passed in nor persisted on the run") judge_model_id = int(judge_model_id) - if JiuwenSDKAdapter is None: + adapter_class = _get_jiuwen_adapter_class() + if adapter_class is None: raise JiuwenSDKUnavailableError("Jiuwen SDK adapter is unavailable. Please install optional dependencies for openjiuwen.") - adapter = JiuwenSDKAdapter(model_id=judge_model_id, tenant_id=tenant_id) + adapter = adapter_class(model_id=judge_model_id, tenant_id=tenant_id) cases = list_agent_evaluation_cases(agent_evaluation_id=agent_evaluation_id, tenant_id=tenant_id, limit=100000, offset=0) scores: List[float] = [] diff --git a/backend/services/agent_runtime/__init__.py b/backend/services/agent_runtime/__init__.py new file mode 100644 index 0000000000..665c79f3ae --- /dev/null +++ b/backend/services/agent_runtime/__init__.py @@ -0,0 +1,6 @@ +"""Agent-level in-process runtime dispatch.""" + +from .execution import AgentRuntimeExecution +from .registry import get_agent_runtime + +__all__ = ["AgentRuntimeExecution", "get_agent_runtime"] diff --git a/backend/services/agent_runtime/base.py b/backend/services/agent_runtime/base.py new file mode 100644 index 0000000000..6d2b6423a7 --- /dev/null +++ b/backend/services/agent_runtime/base.py @@ -0,0 +1,19 @@ +"""Runtime provider protocol.""" + +from collections.abc import AsyncIterator +from typing import Protocol + +from .execution import AgentRuntimeExecution + + +class AgentRuntime(Protocol): + """Common lifecycle implemented by both in-process frameworks.""" + + async def run(self, execution: AgentRuntimeExecution) -> AsyncIterator[str]: + """Yield legacy Nexent event chunks for one execution.""" + + def request_stop(self, run_id: str) -> bool: + """Signal one active run without blocking the HTTP stop path.""" + + async def shutdown(self) -> None: + """Drain resources owned by an initialized provider.""" diff --git a/backend/services/agent_runtime/execution.py b/backend/services/agent_runtime/execution.py new file mode 100644 index 0000000000..fb6a027b06 --- /dev/null +++ b/backend/services/agent_runtime/execution.py @@ -0,0 +1,19 @@ +"""Framework-neutral execution context passed to in-process runtimes.""" + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class AgentRuntimeExecution: + """All request identity and assembled resources required by a runtime.""" + + run_id: str + agent_run_info: Any + conversation_id: int + user_id: str + tenant_id: str + version_no: int + + +__all__ = ["AgentRuntimeExecution"] diff --git a/backend/services/agent_runtime/openjiuwen_spec.py b/backend/services/agent_runtime/openjiuwen_spec.py new file mode 100644 index 0000000000..13d316d7b7 --- /dev/null +++ b/backend/services/agent_runtime/openjiuwen_spec.py @@ -0,0 +1,51 @@ +"""Recursive OpenJiuwen run specification built from assembled AgentConfig.""" + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class OpenJiuwenRunSpec: + """One Agent node and its same-framework child tree.""" + + agent_id: int + name: str + description: str + agent_config: Any + parent_agent_id: int | None + depth: int + children: tuple["OpenJiuwenRunSpec", ...] + + +def build_openjiuwen_run_spec(agent_config: Any) -> OpenJiuwenRunSpec: + """Validate one same-framework acyclic tree before creating native resources.""" + + def build(config: Any, parent_id: int | None, depth: int, ancestry: tuple[int, ...]): + agent_id = getattr(config, "id", None) + if agent_id is None: + raise ValueError("OpenJiuwen AgentConfig requires a persisted Agent ID.") + if agent_id in ancestry: + raise ValueError(f"Circular internal Agent relationship detected at Agent {agent_id}.") + framework = getattr(config, "runtime_framework", None) + if framework != "openjiuwen": + raise ValueError( + f"OpenJiuwen run tree contains Agent {agent_id} with framework {framework!r}." + ) + children = tuple( + build(child, agent_id, depth + 1, (*ancestry, agent_id)) + for child in getattr(config, "managed_agents", []) + ) + return OpenJiuwenRunSpec( + agent_id=agent_id, + name=config.name, + description=config.description, + agent_config=config, + parent_agent_id=parent_id, + depth=depth, + children=children, + ) + + return build(agent_config, None, 0, ()) + + +__all__ = ["OpenJiuwenRunSpec", "build_openjiuwen_run_spec"] diff --git a/backend/services/agent_runtime/providers/__init__.py b/backend/services/agent_runtime/providers/__init__.py new file mode 100644 index 0000000000..bd40a92811 --- /dev/null +++ b/backend/services/agent_runtime/providers/__init__.py @@ -0,0 +1 @@ +"""Lazy runtime provider implementations.""" diff --git a/backend/services/agent_runtime/providers/openjiuwen_in_process.py b/backend/services/agent_runtime/providers/openjiuwen_in_process.py new file mode 100644 index 0000000000..8f58e4cd7d --- /dev/null +++ b/backend/services/agent_runtime/providers/openjiuwen_in_process.py @@ -0,0 +1,1057 @@ +"""Lazy, request-scoped OpenJiuwen execution inside nexent-runtime.""" + +from __future__ import annotations + +import asyncio +import inspect +import json +import logging +import uuid +from collections.abc import AsyncIterator, Mapping, Sequence +from contextlib import suppress +from dataclasses import dataclass +from threading import RLock +from typing import Any + +from nexent.core.agents.a2a_agent_proxy import ExternalA2AAgentWrapper +from nexent.core.agents.nexent_agent import NexentAgent +from nexent.core.utils.observer import MessageObserver + +from ..execution import AgentRuntimeExecution +from ..openjiuwen_spec import ( + OpenJiuwenRunSpec, + build_openjiuwen_run_spec, +) + +logger = logging.getLogger(__name__) + +_END = object() + + +@dataclass(frozen=True) +class _Failure: + error: Exception + + +@dataclass +class _ActiveRun: + execution: AgentRuntimeExecution + cancel_event: asyncio.Event + task: asyncio.Task[None] + queue: asyncio.Queue[Any] + loop: asyncio.AbstractEventLoop + + +@dataclass(frozen=True) +class _OpenJiuwenBindings: + Runner: Any + ReActAgent: Any + ReActAgentConfig: Any + AgentCard: Any + LocalFunction: Any + ToolCard: Any + McpServerConfig: Any + ModelClientConfig: Any + ModelRequestConfig: Any + ContextEngineConfig: Any + create_agent_session: Any + UserMessage: Any + AssistantMessage: Any + SystemMessage: Any + AgentCallbackEvent: Any + ToolCallInputs: Any + + +def _load_openjiuwen_bindings() -> _OpenJiuwenBindings: + """Import OpenJiuwen only after an OpenJiuwen Agent is selected.""" + from openjiuwen.core.context_engine import ContextEngineConfig + from openjiuwen.core.foundation.llm import AssistantMessage, SystemMessage, UserMessage + from openjiuwen.core.foundation.llm.schema.config import ModelClientConfig, ModelRequestConfig + from openjiuwen.core.foundation.tool import LocalFunction, McpServerConfig, ToolCard + from openjiuwen.core.runner import Runner + from openjiuwen.core.session.agent import create_agent_session + from openjiuwen.core.single_agent import AgentCard, ReActAgent, ReActAgentConfig + from openjiuwen.core.single_agent.rail.base import AgentCallbackEvent, ToolCallInputs + + return _OpenJiuwenBindings( + Runner=Runner, + ReActAgent=ReActAgent, + ReActAgentConfig=ReActAgentConfig, + AgentCard=AgentCard, + LocalFunction=LocalFunction, + ToolCard=ToolCard, + McpServerConfig=McpServerConfig, + ModelClientConfig=ModelClientConfig, + ModelRequestConfig=ModelRequestConfig, + ContextEngineConfig=ContextEngineConfig, + create_agent_session=create_agent_session, + UserMessage=UserMessage, + AssistantMessage=AssistantMessage, + SystemMessage=SystemMessage, + AgentCallbackEvent=AgentCallbackEvent, + ToolCallInputs=ToolCallInputs, + ) + + +def _json_safe(value: Any) -> Any: + if value is None or isinstance(value, (bool, float, int, str)): + return value + if hasattr(value, "model_dump"): + return _json_safe(value.model_dump(mode="json", exclude_none=True)) + if isinstance(value, Mapping): + return {str(key): _json_safe(item) for key, item in value.items()} + if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + return [_json_safe(item) for item in value] + return str(value) + + +def _json_text(value: Any) -> str: + if isinstance(value, str): + return value + return json.dumps(_json_safe(value), ensure_ascii=False) + + +class _EventEmitter: + """Serialize node events and assign one sequence across the recursive tree.""" + + def __init__(self, queue: asyncio.Queue[Any]) -> None: + self.queue = queue + self._lock = asyncio.Lock() + self._sequence = 0 + + async def emit( + self, + event_type: str, + content: Any, + spec: OpenJiuwenRunSpec, + *, + event_kind: str | None = None, + ) -> None: + async with self._lock: + self._sequence += 1 + payload = { + "type": event_type, + "content": _json_text(content), + "sequence": self._sequence, + "runtime_framework": "openjiuwen", + "agent_id": spec.agent_id, + "agent_name": spec.name, + "parent_agent_id": spec.parent_agent_id, + "depth": spec.depth, + } + if event_kind: + payload["runtime_event"] = event_kind + await self.queue.put(json.dumps(payload, ensure_ascii=False)) + + async def emit_legacy_chunk(self, chunk: str, spec: OpenJiuwenRunSpec) -> None: + try: + payload = json.loads(chunk) + except (TypeError, json.JSONDecodeError): + await self.emit("other", str(chunk), spec, event_kind="tool_event") + return + async with self._lock: + self._sequence += 1 + payload.update( + { + "sequence": self._sequence, + "runtime_framework": "openjiuwen", + "agent_id": spec.agent_id, + "agent_name": spec.name, + "parent_agent_id": spec.parent_agent_id, + "depth": spec.depth, + } + ) + await self.queue.put(json.dumps(payload, ensure_ascii=False)) + + +class _NodeScope: + """Own one node invocation's native Agent, tools, MCP clients, and context.""" + + def __init__( + self, + *, + runtime: "OpenJiuwenInProcessRuntime", + execution: AgentRuntimeExecution, + spec: OpenJiuwenRunSpec, + emitter: _EventEmitter, + cancel_event: asyncio.Event, + scope_id: str, + ) -> None: + self.runtime = runtime + self.execution = execution + self.spec = spec + self.emitter = emitter + self.cancel_event = cancel_event + self.scope_id = scope_id + self.agent: Any | None = None + self.session: Any | None = None + self.session_id: str | None = None + self.mcp_server_ids: list[str] = [] + self.tool_instances: list[Any] = [] + self.node_observer = MessageObserver( + lang=getattr(execution.agent_run_info.observer, "lang", "zh") + ) + + async def setup(self) -> None: + bindings = self.runtime._bindings + assert bindings is not None + card = bindings.AgentCard( + id=f"nexent-agent-{self.scope_id}", + name=self.spec.name, + description=self.spec.description, + ) + self.agent = bindings.ReActAgent(card) + self.agent.configure(self.runtime._build_agent_config(self.execution, self.spec)) + await self.runtime._register_callbacks(self, bindings) + await self._setup_local_tools(bindings) + await self._setup_a2a_tools(bindings) + await self._setup_child_tools(bindings) + await self._setup_mcp_tools(bindings) + + self.session_id = f"nexent-session-{self.scope_id}" + self.session = bindings.create_agent_session( + session_id=self.session_id, + card=card, + ) + history = self.runtime._history_messages( + self.execution.agent_run_info.history if self.spec.depth == 0 else None, + bindings, + ) + await self.agent.context_engine.create_context( + session=self.session, + history_messages=history, + ) + + async def _setup_local_tools(self, bindings: _OpenJiuwenBindings) -> None: + tool_factory = NexentAgent( + observer=self.node_observer, + model_config_list=self.execution.agent_run_info.model_config_list, + stop_event=self.execution.agent_run_info.stop_event, + ) + for tool_config in self.spec.agent_config.tools: + if tool_config.source == "mcp": + continue + tool_instance, callable_obj = self._create_request_tool( + tool_factory, + tool_config, + ) + self.tool_instances.append(tool_instance) + card = bindings.ToolCard( + id=f"nexent-tool-{self.scope_id}-{tool_config.name or tool_config.class_name}", + name=tool_config.name or tool_config.class_name, + description=tool_config.description or "", + input_params=self.runtime._resolve_local_tool_input_schema(tool_config), + ) + + async def invoke_local(_callable=callable_obj, **kwargs): + if self.cancel_event.is_set(): + raise asyncio.CancelledError + if inspect.iscoroutinefunction(_callable): + result = await _callable(**kwargs) + else: + result = await asyncio.to_thread(_callable, **kwargs) + if inspect.isawaitable(result): + result = await result + await self._drain_tool_observer() + return _json_safe(result) + + local_function = bindings.LocalFunction(card=card, func=invoke_local) + result = self.agent.ability_manager.add_ability(card, local_function) + if not result.added: + raise ValueError(f"Duplicate OpenJiuwen tool ability: {card.name}") + + @staticmethod + def _create_request_tool(tool_factory: NexentAgent, tool_config: Any) -> tuple[Any, Any]: + """Create a node-owned tool and avoid process-global builtin Skill wrappers.""" + if tool_config.source != "builtin": + instance = tool_factory.create_tool(tool_config) + return instance, getattr(instance, "forward", instance) + + metadata = tool_config.metadata or {} + common_kwargs = { + "local_skills_dir": (tool_config.params or {}).get("local_skills_dir"), + "agent_id": metadata.get("agent_id"), + "tenant_id": metadata.get("tenant_id"), + "version_no": metadata.get("version_no", 0), + } + if tool_config.class_name == "RunSkillScriptTool": + from nexent.core.tools.run_skill_script_tool import RunSkillScriptTool + + instance = RunSkillScriptTool(**common_kwargs) + return instance, instance.execute + if tool_config.class_name == "ReadSkillMdTool": + from nexent.core.tools.read_skill_md_tool import ReadSkillMdTool + + instance = ReadSkillMdTool(**common_kwargs) + + def read_skill_md(skill_name: str, additional_files: list[str] | None = None): + return instance.execute(skill_name, *(additional_files or [])) + + return instance, read_skill_md + if tool_config.class_name == "WriteSkillFileTool": + from nexent.core.tools.write_skill_file_tool import WriteSkillFileTool + + instance = WriteSkillFileTool(**common_kwargs) + return instance, instance.execute + if tool_config.class_name == "ReadSkillConfigTool": + from nexent.core.tools.read_skill_config_tool import ReadSkillConfigTool + + instance = ReadSkillConfigTool(**common_kwargs) + return instance, instance.execute + + instance = tool_factory.create_tool(tool_config) + return instance, getattr(instance, "forward", instance) + + async def _setup_a2a_tools(self, bindings: _OpenJiuwenBindings) -> None: + for external_config in self.spec.agent_config.external_a2a_agents: + wrapper = ExternalA2AAgentWrapper( + agent_info=external_config.to_a2a_agent_info(), + stop_event=self.execution.agent_run_info.stop_event, + observer=self.node_observer, + ) + self.tool_instances.append(wrapper) + card = bindings.ToolCard( + id=f"nexent-a2a-{self.scope_id}-{external_config.agent_id}", + name=external_config.name, + description=external_config.description or "External A2A Agent", + input_params={ + "type": "object", + "properties": {"task": {"type": "string"}}, + "required": ["task"], + }, + ) + + async def invoke_a2a(task: str, _wrapper=wrapper, **kwargs): + if self.cancel_event.is_set(): + raise asyncio.CancelledError + return await asyncio.to_thread(_wrapper.run, task=task, **kwargs) + + local_function = bindings.LocalFunction(card=card, func=invoke_a2a) + result = self.agent.ability_manager.add_ability(card, local_function) + if not result.added: + raise ValueError(f"Duplicate OpenJiuwen A2A ability: {card.name}") + + async def _setup_child_tools(self, bindings: _OpenJiuwenBindings) -> None: + for child_spec in self.spec.children: + card = bindings.ToolCard( + id=f"nexent-child-{self.scope_id}-{child_spec.agent_id}", + name=child_spec.name, + description=child_spec.description, + input_params={ + "type": "object", + "properties": { + "task": { + "type": "string", + "description": "Task delegated to the child Agent.", + } + }, + "required": ["task"], + }, + ) + + async def invoke_child(task: str, _child=child_spec, **_kwargs): + if self.cancel_event.is_set(): + raise asyncio.CancelledError + return await self.runtime._execute_node( + execution=self.execution, + spec=_child, + query=task, + emitter=self.emitter, + cancel_event=self.cancel_event, + ) + + local_function = bindings.LocalFunction(card=card, func=invoke_child) + result = self.agent.ability_manager.add_ability(card, local_function) + if not result.added: + raise ValueError(f"Duplicate OpenJiuwen child Agent ability: {card.name}") + + async def _setup_mcp_tools(self, bindings: _OpenJiuwenBindings) -> None: + configured_mcp_tools = { + str(tool.class_name or tool.name) + for tool in self.spec.agent_config.tools + if tool.source == "mcp" and (tool.class_name or tool.name) + } + bound_mcp_tools = { + tool_name + for binding in self.spec.agent_config.mcp_bindings + for tool_name in binding.tool_names + } + missing_bindings = configured_mcp_tools - bound_mcp_tools + if missing_bindings: + raise RuntimeError( + "Required MCP bindings are unavailable: " + + ", ".join(sorted(missing_bindings)) + ) + + for binding in self.spec.agent_config.mcp_bindings: + if not binding.available: + if binding.required: + raise RuntimeError( + f"Required MCP server is unavailable: {binding.server_name} " + f"({binding.unavailable_reason or 'unavailable'})" + ) + await self.emitter.emit( + "other", + { + "warning": "optional_mcp_unavailable", + "server": binding.server_name, + "reason": binding.unavailable_reason or "unavailable", + }, + self.spec, + event_kind="warning", + ) + continue + server_id = f"nexent-mcp-{self.scope_id}-{binding.server_id}" + config = bindings.McpServerConfig( + server_id=server_id, + server_name=f"{binding.server_name}-{self.scope_id}", + server_path=binding.url, + client_type=binding.transport, + auth_headers=dict(binding.headers), + ) + self.mcp_server_ids.append(server_id) + result = await bindings.Runner.resource_mgr.add_mcp_server( + config, + tag=["nexent", f"nexent-run-{self.execution.run_id}"], + ) + if getattr(result, "is_err", lambda: False)(): + if binding.required: + raise RuntimeError(f"Required MCP server is unavailable: {binding.server_name}") + await self.emitter.emit( + "other", + {"warning": "optional_mcp_unavailable", "server": binding.server_name}, + self.spec, + event_kind="warning", + ) + continue + infos = await bindings.Runner.resource_mgr.get_mcp_tool_infos(server_id=server_id) + info_items = infos if isinstance(infos, list) else [infos] + discovered_names = { + item.name + for item in info_items + if item is not None and getattr(item, "name", None) + } + missing_tools = set(binding.tool_names) - discovered_names + missing_required_tools = set(binding.required_tool_names) - discovered_names + if missing_required_tools: + raise RuntimeError( + f"Required MCP tools are unavailable on {binding.server_name}: " + + ", ".join(sorted(missing_required_tools)) + ) + if missing_tools: + await self.emitter.emit( + "other", + { + "warning": "optional_mcp_tools_unavailable", + "server": binding.server_name, + "tools": sorted(missing_tools), + }, + self.spec, + event_kind="warning", + ) + for tool_name in binding.tool_names: + if tool_name not in discovered_names: + continue + tool = await bindings.Runner.resource_mgr.get_mcp_tool( + name=tool_name, + server_id=server_id, + ) + if isinstance(tool, list): + tool = next((item for item in tool if item is not None), None) + if tool is None: + if tool_name in binding.required_tool_names: + raise RuntimeError(f"Required MCP tool cannot be bound: {tool_name}") + continue + add_result = self.agent.ability_manager.add(tool.card) + if not add_result.added: + raise ValueError(f"Duplicate OpenJiuwen MCP ability: {tool_name}") + + async def _drain_tool_observer(self) -> None: + for chunk in self.node_observer.get_cached_message(): + await self.emitter.emit_legacy_chunk(chunk, self.spec) + + async def cleanup(self) -> None: + bindings = self.runtime._bindings + if bindings is None: + return + if self.agent is not None: + with suppress(Exception): + await self.agent.agent_callback_manager.clear() + with suppress(Exception): + self.agent.ability_manager.teardown_tools() + for server_id in reversed(self.mcp_server_ids): + with suppress(Exception): + await bindings.Runner.resource_mgr.remove_mcp_server( + server_id=server_id, + skip_if_tag_not_exists=True, + ignore_exception=True, + ) + if self.agent is not None and self.session_id is not None: + with suppress(Exception): + await self.agent.context_engine.clear_context(session_id=self.session_id) + for tool_instance in reversed(self.tool_instances): + close = getattr(tool_instance, "aclose", None) + if callable(close): + with suppress(Exception): + await close() + continue + close = getattr(tool_instance, "close", None) + if callable(close): + with suppress(Exception): + result = close() + if inspect.isawaitable(result): + await result + self.mcp_server_ids.clear() + self.tool_instances.clear() + + +class OpenJiuwenInProcessRuntime: + """Own the lazily started Runner and all active OpenJiuwen run tasks.""" + + def __init__(self) -> None: + self._bindings: _OpenJiuwenBindings | None = None + self._start_lock: asyncio.Lock | None = None + self._started = False + self._active: dict[str, _ActiveRun] = {} + self._active_lock = RLock() + self._shutting_down = False + + async def _ensure_started(self) -> None: + if self._started: + return + if self._start_lock is None: + self._start_lock = asyncio.Lock() + async with self._start_lock: + if self._shutting_down: + raise RuntimeError("OpenJiuwen runtime is shutting down.") + if self._started: + return + self._bindings = _load_openjiuwen_bindings() + await self._bindings.Runner.start() + self._started = True + + async def run(self, execution: AgentRuntimeExecution) -> AsyncIterator[str]: + if self._shutting_down: + raise RuntimeError("OpenJiuwen runtime is shutting down.") + try: + await self._ensure_started() + except Exception as exc: + fallback_spec = OpenJiuwenRunSpec( + agent_id=getattr(execution.agent_run_info.agent_config, "id", -1), + name=getattr(execution.agent_run_info.agent_config, "name", "openjiuwen"), + description="", + agent_config=execution.agent_run_info.agent_config, + parent_agent_id=None, + depth=0, + children=(), + ) + queue: asyncio.Queue[Any] = asyncio.Queue() + await _EventEmitter(queue).emit( + "error", + "OpenJiuwen runtime initialization failed.", + fallback_spec, + event_kind="initialization_error", + ) + yield await queue.get() + raise RuntimeError("OpenJiuwen runtime initialization failed.") from exc + if self._shutting_down: + raise RuntimeError("OpenJiuwen runtime is shutting down.") + queue: asyncio.Queue[Any] = asyncio.Queue() + cancel_event = asyncio.Event() + emitter = _EventEmitter(queue) + task = asyncio.create_task( + self._produce(execution, cancel_event, emitter, queue), + name=f"openjiuwen-run-{execution.run_id}", + ) + active = _ActiveRun( + execution=execution, + cancel_event=cancel_event, + task=task, + queue=queue, + loop=asyncio.get_running_loop(), + ) + with self._active_lock: + if execution.run_id in self._active: + task.cancel() + raise ValueError(f"OpenJiuwen run already exists: {execution.run_id}") + self._active[execution.run_id] = active + try: + while True: + item = await queue.get() + if item is _END: + break + if isinstance(item, _Failure): + raise RuntimeError("OpenJiuwen execution failed.") from item.error + yield item + try: + await task + except asyncio.CancelledError: + if not cancel_event.is_set(): + raise + finally: + cancel_event.set() + if not task.done(): + task.cancel() + with suppress(asyncio.CancelledError): + await task + with self._active_lock: + current = self._active.get(execution.run_id) + if current is active: + self._active.pop(execution.run_id, None) + + async def _produce( + self, + execution: AgentRuntimeExecution, + cancel_event: asyncio.Event, + emitter: _EventEmitter, + queue: asyncio.Queue[Any], + ) -> None: + try: + spec = build_openjiuwen_run_spec(execution.agent_run_info.agent_config) + await self._execute_node( + execution=execution, + spec=spec, + query=execution.agent_run_info.query, + emitter=emitter, + cancel_event=cancel_event, + ) + except asyncio.CancelledError: + execution.agent_run_info.stop_event.set() + except asyncio.TimeoutError as exc: + logger.exception("OpenJiuwen in-process run timed out, run_id=%s", execution.run_id) + await emitter.emit( + "error", + "OpenJiuwen execution timed out.", + spec if "spec" in locals() else OpenJiuwenRunSpec( + agent_id=getattr(execution.agent_run_info.agent_config, "id", -1), + name=getattr(execution.agent_run_info.agent_config, "name", "openjiuwen"), + description="", + agent_config=execution.agent_run_info.agent_config, + parent_agent_id=None, + depth=0, + children=(), + ), + event_kind="timeout", + ) + await queue.put(_Failure(exc)) + except Exception as exc: + logger.exception("OpenJiuwen in-process run failed, run_id=%s", execution.run_id) + await emitter.emit( + "error", + "OpenJiuwen execution failed.", + spec if "spec" in locals() else OpenJiuwenRunSpec( + agent_id=getattr(execution.agent_run_info.agent_config, "id", -1), + name=getattr(execution.agent_run_info.agent_config, "name", "openjiuwen"), + description="", + agent_config=execution.agent_run_info.agent_config, + parent_agent_id=None, + depth=0, + children=(), + ), + event_kind="error", + ) + await queue.put(_Failure(exc)) + finally: + await queue.put(_END) + + async def _execute_node( + self, + *, + execution: AgentRuntimeExecution, + spec: OpenJiuwenRunSpec, + query: str, + emitter: _EventEmitter, + cancel_event: asyncio.Event, + ) -> str: + if cancel_event.is_set(): + raise asyncio.CancelledError + scope_id = f"{execution.run_id}-{spec.agent_id}-{uuid.uuid4().hex[:8]}" + scope = _NodeScope( + runtime=self, + execution=execution, + spec=spec, + emitter=emitter, + cancel_event=cancel_event, + scope_id=scope_id, + ) + output_parts: list[str] = [] + final_answer = "" + try: + await emitter.emit( + "agent_new_run", + query, + spec, + event_kind="agent_started", + ) + await scope.setup() + stream = scope.agent.stream( + {"query": query, "conversation_id": scope.session.get_session_id()}, + session=scope.session, + ) + try: + async for chunk in stream: + if cancel_event.is_set(): + raise asyncio.CancelledError + chunk_type = str(getattr(chunk, "type", "")) + payload = getattr(chunk, "payload", None) + payload = payload if isinstance(payload, Mapping) else {"content": payload} + if chunk_type == "llm_reasoning": + await emitter.emit( + "model_output_deep_thinking", + payload.get("content", ""), + spec, + event_kind="model_reasoning_delta", + ) + elif chunk_type == "llm_output": + content = str(payload.get("content") or "") + output_parts.append(content) + await emitter.emit( + "model_output_thinking", + content, + spec, + event_kind="model_output_delta", + ) + elif chunk_type == "llm_usage": + await emitter.emit( + "token_count", + payload.get("usage_metadata", payload), + spec, + event_kind="token_usage", + ) + elif chunk_type == "answer": + if str(payload.get("result_type") or "answer") == "error": + raise RuntimeError("OpenJiuwen Agent returned an error result.") + final_answer = str(payload.get("output") or payload.get("content") or "") + finally: + close_stream = getattr(stream, "aclose", None) + if callable(close_stream): + with suppress(asyncio.CancelledError, Exception): + await close_stream() + if not final_answer: + final_answer = "".join(output_parts) + if spec.depth == 0: + await emitter.emit( + "final_answer", + final_answer, + spec, + event_kind="final_answer", + ) + else: + await emitter.emit( + "agent_finish", + final_answer, + spec, + event_kind="child_agent_finished", + ) + return final_answer + finally: + await scope.cleanup() + + def _build_agent_config( + self, + execution: AgentRuntimeExecution, + spec: OpenJiuwenRunSpec, + ) -> Any: + bindings = self._bindings + assert bindings is not None + model = next( + ( + item + for item in execution.agent_run_info.model_config_list + if item is not None and item.cite_name == spec.agent_config.model_name + ), + None, + ) + if model is None: + raise ValueError(f"OpenJiuwen model config not found: {spec.agent_config.model_name}") + provider = self._model_provider(model.model_factory) + ssl_cert = getattr(model, "ssl_cert", None) + verify_ssl = bool(model.ssl_verify) and bool(ssl_cert) + client_kwargs = { + "client_id": f"nexent-model-{execution.run_id}-{spec.agent_id}", + "client_provider": provider, + "api_key": model.api_key, + "api_base": model.url, + "verify_ssl": verify_ssl, + } + if ssl_cert: + client_kwargs["ssl_cert"] = ssl_cert + if model.timeout_seconds: + client_kwargs["timeout"] = model.timeout_seconds + model_client_config = bindings.ModelClientConfig(**client_kwargs) + request_kwargs = { + "model": model.model_name, + "temperature": model.temperature, + "top_p": model.top_p, + "max_tokens": model.max_output_tokens, + } + if model.extra_body: + request_kwargs["extra_body"] = model.extra_body + model_request_config = bindings.ModelRequestConfig(**request_kwargs) + context_config = bindings.ContextEngineConfig( + max_context_message_num=100, + context_window_tokens=model.context_window_tokens, + model_name=model.model_name, + ) + return bindings.ReActAgentConfig( + model_name=model.model_name, + model_provider=provider, + api_key=model.api_key, + api_base=model.url, + max_iterations=spec.agent_config.max_steps, + model_client_config=model_client_config, + model_config_obj=model_request_config, + context_engine_config=context_config, + prompt_template=[ + {"role": "system", "content": self._prompt_text(spec.agent_config)} + ], + ) + + @staticmethod + def _model_provider(model_factory: str | None) -> str: + normalized = str(model_factory or "").lower() + if "anthropic" in normalized: + return "Anthropic" + if "openrouter" in normalized: + return "OpenRouter" + if "silicon" in normalized: + return "SiliconFlow" + if "dashscope" in normalized or "qwen" in normalized: + return "DashScope" + if "deepseek" in normalized: + return "DeepSeek" + return "OpenAI" + + @staticmethod + def _prompt_text(agent_config: Any) -> str: + sections: list[str] = [] + for component in getattr(agent_config, "context_components", None) or []: + for message in component.to_messages(): + content = message.get("content", "") + if isinstance(content, list): + sections.extend( + str(part.get("text") or "") + for part in content + if isinstance(part, dict) and part.get("text") + ) + elif content: + sections.append(str(content)) + if not sections: + prompt_templates = getattr(agent_config, "prompt_templates", None) or {} + system_prompt = prompt_templates.get("system_prompt") + if system_prompt: + sections.append(str(system_prompt)) + instructions = getattr(agent_config, "instructions", None) + if instructions: + sections.insert(0, str(instructions)) + return "\n\n".join(section for section in sections if section) + + @staticmethod + def _history_messages(history: Any, bindings: _OpenJiuwenBindings) -> list[Any]: + messages = [] + for item in history or []: + common = {"content": item.content} + if item.role == "assistant": + messages.append(bindings.AssistantMessage(**common)) + elif item.role == "system": + messages.append(bindings.SystemMessage(**common)) + else: + messages.append(bindings.UserMessage(**common)) + return messages + + @staticmethod + def _tool_input_schema(raw_inputs: str | None) -> dict[str, Any]: + if not raw_inputs: + return {"type": "object", "properties": {}} + try: + parsed = json.loads(raw_inputs) if isinstance(raw_inputs, str) else raw_inputs + except json.JSONDecodeError: + return {"type": "object", "properties": {}} + if not isinstance(parsed, dict): + return {"type": "object", "properties": {}} + if parsed.get("type") == "object" and isinstance(parsed.get("properties"), dict): + return parsed + + def normalize_property(value: Any) -> dict[str, Any]: + if isinstance(value, dict): + return value + type_name = str(value or "string").strip().lower().replace("typing.", "") + optional = "optional[" in type_name or "none" in type_name + if "list" in type_name or type_name.endswith("[]"): + item_type = "string" + if "int" in type_name: + item_type = "integer" + elif "float" in type_name or "number" in type_name: + item_type = "number" + schema = {"type": "array", "items": {"type": item_type}} + elif "dict" in type_name or "object" in type_name: + schema = {"type": "object"} + elif "bool" in type_name: + schema = {"type": "boolean"} + elif "int" in type_name: + schema = {"type": "integer"} + elif "float" in type_name or "number" in type_name: + schema = {"type": "number"} + else: + schema = {"type": "string"} + if optional: + schema["nullable"] = True + return schema + + properties = { + str(name): normalize_property(value) + for name, value in parsed.items() + } + required = [ + name + for name, value in properties.items() + if value.get("nullable") is not True and "default" not in value + ] + schema: dict[str, Any] = {"type": "object", "properties": properties} + if required: + schema["required"] = required + return schema + + @classmethod + def _resolve_local_tool_input_schema(cls, tool_config: Any) -> dict[str, Any]: + """Resolve local tool schemas without changing shared tool metadata.""" + if tool_config.class_name == "RunSkillScriptTool": + return { + "type": "object", + "properties": { + "skill_name": {"type": "string"}, + "script_path": {"type": "string"}, + "params": {"type": "string"}, + }, + "required": ["skill_name", "script_path"], + "additionalProperties": False, + } + if tool_config.class_name == "ReadSkillMdTool": + return { + "type": "object", + "properties": { + "skill_name": {"type": "string"}, + "additional_files": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["skill_name"], + "additionalProperties": False, + } + return cls._tool_input_schema(tool_config.inputs) + + async def _register_callbacks( + self, + scope: _NodeScope, + bindings: _OpenJiuwenBindings, + ) -> None: + step = 0 + + async def before_model(_context): + nonlocal step + step += 1 + await scope.emitter.emit( + "step_count", + step, + scope.spec, + event_kind="step_started", + ) + + async def model_error(_context): + await scope.emitter.emit( + "error", + "OpenJiuwen model call failed.", + scope.spec, + event_kind="model_error", + ) + + async def before_tool(context): + inputs = context.inputs + tool_name = getattr(inputs, "tool_name", "") + await scope.emitter.emit( + "tool", + { + "event": "started", + "tool_name": tool_name, + "arguments": _json_safe(getattr(inputs, "tool_args", None)), + }, + scope.spec, + event_kind="tool_call_started", + ) + + async def after_tool(context): + inputs = context.inputs + await scope.emitter.emit( + "execution_logs", + { + "tool_name": getattr(inputs, "tool_name", ""), + "result": _json_safe(getattr(inputs, "tool_result", None)), + }, + scope.spec, + event_kind="tool_call_finished", + ) + + async def tool_error(context): + inputs = context.inputs + await scope.emitter.emit( + "error", + f"OpenJiuwen tool call failed: {getattr(inputs, 'tool_name', '')}", + scope.spec, + event_kind="tool_error", + ) + + callbacks = { + bindings.AgentCallbackEvent.BEFORE_MODEL_CALL: before_model, + bindings.AgentCallbackEvent.ON_MODEL_EXCEPTION: model_error, + bindings.AgentCallbackEvent.BEFORE_TOOL_CALL: before_tool, + bindings.AgentCallbackEvent.AFTER_TOOL_CALL: after_tool, + bindings.AgentCallbackEvent.ON_TOOL_EXCEPTION: tool_error, + } + for event, callback in callbacks.items(): + await scope.agent.register_callback(event, callback) + + def request_stop(self, run_id: str) -> bool: + with self._active_lock: + active = self._active.get(run_id) + if active is None: + return False + active.execution.agent_run_info.stop_event.set() + + def cancel_run() -> None: + active.cancel_event.set() + active.task.cancel() + active.queue.put_nowait(_END) + + active.loop.call_soon_threadsafe(cancel_run) + return True + + async def shutdown(self) -> None: + self._shutting_down = True + if self._start_lock is not None: + async with self._start_lock: + pass + if not self._started or self._bindings is None: + return + with self._active_lock: + active_runs = list(self._active.values()) + for active in active_runs: + active.execution.agent_run_info.stop_event.set() + active.cancel_event.set() + active.task.cancel() + active.queue.put_nowait(_END) + if active_runs: + await asyncio.gather( + *(active.task for active in active_runs), + return_exceptions=True, + ) + await self._bindings.Runner.stop() + self._started = False + + +def create_runtime() -> OpenJiuwenInProcessRuntime: + """Create the lazy in-process OpenJiuwen provider.""" + return OpenJiuwenInProcessRuntime() + + +__all__ = ["OpenJiuwenInProcessRuntime", "create_runtime"] diff --git a/backend/services/agent_runtime/providers/smolagents.py b/backend/services/agent_runtime/providers/smolagents.py new file mode 100644 index 0000000000..4ef3d911c6 --- /dev/null +++ b/backend/services/agent_runtime/providers/smolagents.py @@ -0,0 +1,48 @@ +"""Existing Smolagents execution exposed through the common provider interface.""" + +from collections.abc import AsyncIterator +from threading import RLock + +from nexent.core.agents.run_agent import agent_run + +from ..execution import AgentRuntimeExecution + + +class SmolagentsRuntime: + """Run the existing AgentRunInfo path without importing OpenJiuwen.""" + + def __init__(self) -> None: + self._stop_events: dict[str, object] = {} + self._lock = RLock() + + async def run(self, execution: AgentRuntimeExecution) -> AsyncIterator[str]: + with self._lock: + self._stop_events[execution.run_id] = execution.agent_run_info.stop_event + try: + async for chunk in agent_run(execution.agent_run_info): + yield chunk + finally: + with self._lock: + self._stop_events.pop(execution.run_id, None) + + def request_stop(self, run_id: str) -> bool: + with self._lock: + stop_event = self._stop_events.get(run_id) + if stop_event is None: + return False + stop_event.set() + return True + + async def shutdown(self) -> None: + with self._lock: + stop_events = list(self._stop_events.values()) + for stop_event in stop_events: + stop_event.set() + + +def create_runtime() -> SmolagentsRuntime: + """Create the local Smolagents provider.""" + return SmolagentsRuntime() + + +__all__ = ["SmolagentsRuntime", "create_runtime"] diff --git a/backend/services/agent_runtime/registry.py b/backend/services/agent_runtime/registry.py new file mode 100644 index 0000000000..3067ed39df --- /dev/null +++ b/backend/services/agent_runtime/registry.py @@ -0,0 +1,67 @@ +"""Lazy in-process runtime registry keyed by persisted Agent framework.""" + +import importlib +from threading import RLock +from typing import Any, Callable + +from consts.agent_runtime import normalize_agent_runtime_framework + + +RuntimeFactory = Callable[[], Any] + +_FACTORY_PATHS = { + "smolagents": ".providers.smolagents:create_runtime", + "openjiuwen": ".providers.openjiuwen_in_process:create_runtime", +} +_instances: dict[str, Any] = {} +_lock = RLock() + + +def _load_factory(path: str) -> RuntimeFactory: + module_name, attribute = path.split(":", 1) + module = importlib.import_module(module_name, package=__package__) + factory = getattr(module, attribute) + if not callable(factory): + raise TypeError(f"Runtime factory is not callable: {path}") + return factory + + +def get_agent_runtime(framework: str): + """Return one lazily constructed provider without loading the other framework.""" + normalized = normalize_agent_runtime_framework(framework, default=None) + if normalized is None: + raise ValueError("runtime_framework is required before Agent execution.") + with _lock: + instance = _instances.get(normalized) + if instance is None: + instance = _load_factory(_FACTORY_PATHS[normalized])() + _instances[normalized] = instance + return instance + + +def initialized_runtime_frameworks() -> tuple[str, ...]: + """Return initialized provider names without causing imports.""" + with _lock: + return tuple(sorted(_instances)) + + +async def shutdown_initialized_runtimes() -> None: + """Shutdown only providers that were selected during this process lifetime.""" + with _lock: + instances = list(_instances.values()) + for instance in instances: + await instance.shutdown() + + +def reset_runtime_registry_for_test() -> None: + """Clear cached providers for isolated unit tests.""" + with _lock: + _instances.clear() + + +__all__ = [ + "get_agent_runtime", + "initialized_runtime_frameworks", + "reset_runtime_registry_for_test", + "shutdown_initialized_runtimes", +] diff --git a/backend/services/agent_runtime/run_control.py b/backend/services/agent_runtime/run_control.py new file mode 100644 index 0000000000..ed9f39e9d7 --- /dev/null +++ b/backend/services/agent_runtime/run_control.py @@ -0,0 +1,49 @@ +"""Conversation-to-runtime run lookup used by the existing stop endpoint.""" + +from dataclasses import dataclass +from threading import RLock +from typing import Any + + +@dataclass(frozen=True) +class RuntimeRunHandle: + """One active provider run addressable by conversation and user.""" + + run_id: str + conversation_id: int + user_id: str + runtime: Any + + +class RuntimeRunControlRegistry: + """Track active runs without introducing a second cancel endpoint.""" + + def __init__(self) -> None: + self._by_key: dict[tuple[int, str], RuntimeRunHandle] = {} + self._lock = RLock() + + def register(self, handle: RuntimeRunHandle) -> None: + with self._lock: + self._by_key[(handle.conversation_id, handle.user_id)] = handle + + def unregister(self, *, run_id: str, conversation_id: int, user_id: str) -> None: + key = (conversation_id, user_id) + with self._lock: + current = self._by_key.get(key) + if current is not None and current.run_id == run_id: + self._by_key.pop(key, None) + + def request_stop(self, *, conversation_id: int, user_id: str) -> bool: + with self._lock: + handle = self._by_key.get((conversation_id, user_id)) + return bool(handle and handle.runtime.request_stop(handle.run_id)) + + +runtime_run_control_registry = RuntimeRunControlRegistry() + + +__all__ = [ + "RuntimeRunControlRegistry", + "RuntimeRunHandle", + "runtime_run_control_registry", +] diff --git a/backend/services/agent_service.py b/backend/services/agent_service.py index 7435d74060..48ae59c48e 100644 --- a/backend/services/agent_service.py +++ b/backend/services/agent_service.py @@ -12,7 +12,6 @@ from fastapi import Header, Request from fastapi.responses import JSONResponse, StreamingResponse -from nexent.core.agents.run_agent import agent_run from nexent.memory.memory_service import clear_memory, add_memory_in_levels from jinja2 import Template @@ -24,6 +23,10 @@ from consts.const import MEMORY_SEARCH_START_MSG, MEMORY_SEARCH_DONE_MSG, MEMORY_SEARCH_FAIL_MSG, TOOL_TYPE_MAPPING, \ LANGUAGE, MESSAGE_ROLE, MODEL_CONFIG_MAPPING, CAN_EDIT_ALL_USER_ROLES, PERMISSION_PRIVATE, STREAM_STATUS_EVENT, \ DEFAULT_EN_TITLE, DEFAULT_ZH_TITLE, RUNTIME_CANCEL_POLL_INTERVAL_SECONDS +from consts.agent_runtime import ( + DEFAULT_AGENT_RUNTIME_FRAMEWORK, + normalize_agent_runtime_framework, +) from consts.exceptions import AppException, MemoryPreparationException, SkillDuplicateError from consts.error_code import ErrorCode from consts.agent_unavailable_reasons import AgentUnavailableReason @@ -37,7 +40,6 @@ ExportAndImportDataFormat, MCPInfo, MessageRequest, - MessageUnit, SkillInstanceInfoRequest, SkillZipEntry, ToolInstanceInfoRequest, @@ -70,7 +72,6 @@ delete_tools_by_agent_id, query_all_enabled_tool_instances, query_all_tools, - query_tool_instances_by_id, query_tool_instances_by_agent_id, search_tools_for_sub_agent ) @@ -107,6 +108,12 @@ from services.memory_config_service import build_memory_context from services.streaming_channel import streaming_channel_manager from services.runtime_state_service import runtime_state_service +from .agent_runtime.execution import AgentRuntimeExecution +from .agent_runtime.registry import get_agent_runtime +from .agent_runtime.run_control import ( + RuntimeRunHandle, + runtime_run_control_registry, +) from utils.auth_utils import get_current_user_info, get_user_language from utils.config_utils import tenant_config_manager from utils.memory_utils import build_memory_config @@ -291,7 +298,7 @@ async def _process_skill_file_uploads( absolute_path, error_message, ) - except Exception as exc: + except Exception: logger.exception( "[skill-file] failed to upload file file_name=%s absolute_path=%s", file_name, @@ -967,23 +974,53 @@ async def _stream_agent_chunks( user_id=user_id ) - cancel_poll_task = asyncio.create_task( - _poll_runtime_cancel_signal( + cancel_poll_task: Optional[asyncio.Task] = None + runtime = None + runtime_run_id: Optional[str] = None + try: + runtime_framework = vars(agent_run_info).get( + "runtime_framework", + DEFAULT_AGENT_RUNTIME_FRAMEWORK, + ) + if runtime_framework is None: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED, + "Agent must select a runtime framework before it can run.", + ) + runtime = get_agent_runtime(runtime_framework) + runtime_run_id = uuid.uuid4().hex + runtime_execution = AgentRuntimeExecution( + run_id=runtime_run_id, + agent_run_info=agent_run_info, conversation_id=agent_request.conversation_id, user_id=user_id, - stop_event=agent_run_info.stop_event, + tenant_id=tenant_id, + version_no=agent_request.version_no or 0, + ) + runtime_run_control_registry.register( + RuntimeRunHandle( + run_id=runtime_run_id, + conversation_id=agent_request.conversation_id, + user_id=user_id, + runtime=runtime, + ) + ) + cancel_poll_task = asyncio.create_task( + _poll_runtime_cancel_signal( + conversation_id=agent_request.conversation_id, + user_id=user_id, + stop_event=agent_run_info.stop_event, + ) ) - ) - # In resume mode, emit a status event first - if is_resume_mode: - await channel.publish(STREAM_STATUS_EVENT) - await channel.publish(f'data: {{"status": "resumed", "last_unit_index": {resume_from_unit_index - 1}}}\n\n') - yield STREAM_STATUS_EVENT - yield f'data: {{"status": "resumed", "last_unit_index": {resume_from_unit_index - 1}}}\n\n' + # In resume mode, emit a status event first + if is_resume_mode: + await channel.publish(STREAM_STATUS_EVENT) + await channel.publish(f'data: {{"status": "resumed", "last_unit_index": {resume_from_unit_index - 1}}}\n\n') + yield STREAM_STATUS_EVENT + yield f'data: {{"status": "resumed", "last_unit_index": {resume_from_unit_index - 1}}}\n\n' - try: - async for chunk in agent_run(agent_run_info): + async for chunk in runtime.run(runtime_execution): chunk_type: Optional[str] = None chunk_content: str = "" try: @@ -1099,9 +1136,7 @@ async def _stream_agent_chunks( # loop is async but the DB operations are I/O-bound with network # latency, synchronous writes here are acceptably fast and guarantee # that each chunk is fully persisted before the next chunk arrives. - old_len = len(current_unit["content"]) current_unit["content"] += chunk_content - new_len = len(current_unit["content"]) update_unit_content( current_unit["unit_id"], current_unit["content"], @@ -1292,7 +1327,7 @@ async def _stream_agent_chunks( except Exception: logger.exception("Failed to mark assistant message as %s", terminal_status) - if not cancel_poll_task.done(): + if cancel_poll_task is not None and not cancel_poll_task.done(): cancel_poll_task.cancel() was_stopped = getattr(agent_run_info, "stop_event", None) and agent_run_info.stop_event.is_set() @@ -1301,6 +1336,20 @@ async def _stream_agent_chunks( agent_run_manager.unregister_agent_run( agent_request.conversation_id, user_id, status=terminal_status) + if ( + runtime is not None + and runtime_run_id is not None + and not stream_completed_normally + and not agent_run_info.stop_event.is_set() + ): + runtime.request_stop(runtime_run_id) + if runtime_run_id is not None: + runtime_run_control_registry.unregister( + run_id=runtime_run_id, + conversation_id=agent_request.conversation_id, + user_id=user_id, + ) + # Mark channel as completed and schedule cleanup if channel is not None: await streaming_channel_manager.complete_channel( @@ -1309,7 +1358,7 @@ async def _stream_agent_chunks( status=terminal_status ) # Schedule channel removal (give subscribers time to receive final chunks) - cleanup_task = asyncio.create_task( + asyncio.create_task( _cleanup_channel_later( conversation_id=agent_request.conversation_id, user_id=user_id @@ -1633,9 +1682,119 @@ async def get_creating_sub_agent_info_impl(authorization: str = Header(None)): "duty_prompt": agent_info.get("duty_prompt"), "constraint_prompt": agent_info.get("constraint_prompt"), "few_shots_prompt": agent_info.get("few_shots_prompt"), + "runtime_framework": agent_info.get("runtime_framework"), "sub_agent_id_list": query_sub_agents_id_list(main_agent_id=sub_agent_id, tenant_id=tenant_id)} +def _resolve_runtime_framework_for_save( + request: AgentInfoRequest, + tenant_id: str, +) -> str: + """Resolve the one-time framework assignment and enforce immutability.""" + raw_requested_framework = vars(request).get("runtime_framework") + requested_framework = normalize_agent_runtime_framework( + raw_requested_framework, + default=DEFAULT_AGENT_RUNTIME_FRAMEWORK, + ) + if request.agent_id is None: + return requested_framework + + existing = search_agent_info_by_agent_id( + agent_id=request.agent_id, + tenant_id=tenant_id, + version_no=vars(request).get("version_no", 0), + ) + existing_framework = _runtime_framework_from_record(existing) + if existing_framework is None: + return requested_framework + if raw_requested_framework is not None and requested_framework != existing_framework: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_IMMUTABLE, + "AGENT_RUNTIME_FRAMEWORK_IMMUTABLE", + details={ + "agent_id": request.agent_id, + "current": existing_framework, + "requested": requested_framework, + }, + ) + return existing_framework + + +def _runtime_framework_from_record(record: Any) -> str | None: + """Read persisted framework while treating pre-migration records as Smolagents.""" + if record is None: + return None + if not isinstance(record, dict) or "runtime_framework" not in record: + return DEFAULT_AGENT_RUNTIME_FRAMEWORK + return normalize_agent_runtime_framework( + record.get("runtime_framework"), + default=None, + ) + + +def _require_agent_runtime_framework_for_run( + *, + agent_id: int, + tenant_id: str, + version_no: int = 0, +) -> str: + """Fail before opening SSE when a blank Agent has no selected framework.""" + agent = search_agent_info_by_agent_id( + agent_id=agent_id, + tenant_id=tenant_id, + version_no=version_no, + ) + framework = _runtime_framework_from_record(agent) + if framework is None: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED, + "Agent must select a runtime framework before it can run.", + details={"agent_id": agent_id}, + ) + return framework + + +def _validate_related_agent_frameworks( + *, + parent_agent_id: int, + child_agent_ids: List[int], + tenant_id: str, + version_no: int = 0, +) -> None: + """Reject internal parent-child links that cross runtime frameworks.""" + parent = search_agent_info_by_agent_id(parent_agent_id, tenant_id, version_no) + parent_framework = _runtime_framework_from_record(parent) + if parent_framework is None: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED, + "Parent Agent must select a runtime framework before adding sub-agents.", + details={"agent_id": parent_agent_id}, + ) + + mismatches = [] + for child_agent_id in child_agent_ids: + child = search_agent_info_by_agent_id(child_agent_id, tenant_id, version_no) + child_framework = _runtime_framework_from_record(child) + if child_framework != parent_framework: + mismatches.append( + { + "agent_id": child_agent_id, + "runtime_framework": child_framework, + } + ) + + if mismatches: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_MISMATCH, + "Internal parent and child Agents must use the same runtime framework.", + details={ + "parent_agent_id": parent_agent_id, + "parent_runtime_framework": parent_framework, + "children": mismatches, + }, + ) + + def _validate_requested_output_tokens_for_agent( request: AgentInfoRequest, tenant_id: str, @@ -1681,6 +1840,7 @@ async def update_agent_info_impl(request: AgentInfoRequest, authorization: str = raise AppException(ErrorCode.COMMON_PARAMETER_INVALID, "example_questions cannot exceed 6 items") _validate_requested_output_tokens_for_agent(request, tenant_id) + request.runtime_framework = _resolve_runtime_framework_for_save(request, tenant_id) prompt_template_id, prompt_template_name = get_prompt_template_summary( template_id=request.prompt_template_id, @@ -1715,6 +1875,7 @@ async def update_agent_info_impl(request: AgentInfoRequest, authorization: str = "few_shots_prompt": request.few_shots_prompt, "greeting_message": request.greeting_message, "example_questions": request.example_questions, + "runtime_framework": request.runtime_framework, "enabled": request.enabled if request.enabled is not None else True, "group_ids": convert_list_to_string(request.group_ids) if request.group_ids else user_group_ids, "ingroup_permission": request.ingroup_permission @@ -1725,6 +1886,8 @@ async def update_agent_info_impl(request: AgentInfoRequest, authorization: str = request.prompt_template_id = prompt_template_id request.prompt_template_name = prompt_template_name update_agent(agent_id, request, user_id) + except AppException: + raise except Exception as e: logger.error(f"Failed to update agent info: {str(e)}") raise ValueError(f"Failed to update agent info: {str(e)}") @@ -1835,6 +1998,12 @@ async def update_agent_info_impl(request: AgentInfoRequest, authorization: str = try: if request.related_agent_ids is not None and agent_id is not None: related_agent_ids = request.related_agent_ids + _validate_related_agent_frameworks( + parent_agent_id=agent_id, + child_agent_ids=related_agent_ids, + tenant_id=tenant_id, + version_no=vars(request).get("version_no", 0), + ) # Check for circular dependencies using BFS search_list = deque(related_agent_ids) agent_id_set = set() @@ -1859,7 +2028,7 @@ async def update_agent_info_impl(request: AgentInfoRequest, authorization: str = tenant_id=tenant_id, user_id=user_id ) - except ValueError as e: + except (AppException, ValueError): # Re-raise ValueError (circular dependency) as-is raise except Exception as e: @@ -2258,10 +2427,70 @@ async def export_agent_by_agent_id( skill_names=skill_names, prompt_template_id=agent_info.get( "prompt_template_id"), - prompt_template_name=agent_info.get("prompt_template_name")) + prompt_template_name=agent_info.get("prompt_template_name"), + runtime_framework=normalize_agent_runtime_framework( + agent_info.get("runtime_framework") + )) return agent_info +def _validate_import_agent_graph(agent_bundle: ExportAndImportDataFormat) -> None: + """Validate framework compatibility and acyclicity before import writes begin.""" + records = {int(agent_id): info for agent_id, info in agent_bundle.agent_info.items()} + if agent_bundle.agent_id not in records: + raise AppException( + ErrorCode.COMMON_PARAMETER_INVALID, + "The imported root Agent is missing from agent_info.", + ) + + frameworks = { + agent_id: normalize_agent_runtime_framework( + vars(info).get("runtime_framework") + ) + for agent_id, info in records.items() + } + for parent_id, info in records.items(): + for child_id in info.managed_agents: + if child_id not in records: + raise AppException( + ErrorCode.COMMON_PARAMETER_INVALID, + f"Imported child Agent {child_id} is missing from agent_info.", + details={"parent_agent_id": parent_id, "child_agent_id": child_id}, + ) + if frameworks[child_id] != frameworks[parent_id]: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_MISMATCH, + "Imported internal parent and child Agents must use the same runtime framework.", + details={ + "parent_agent_id": parent_id, + "parent_runtime_framework": frameworks[parent_id], + "child_agent_id": child_id, + "child_runtime_framework": frameworks[child_id], + }, + ) + + visiting: set[int] = set() + visited: set[int] = set() + + def visit(agent_id: int) -> None: + if agent_id in visiting: + raise AppException( + ErrorCode.COMMON_PARAMETER_INVALID, + "Circular dependency detected in imported Agent relationships.", + details={"agent_id": agent_id}, + ) + if agent_id in visited: + return + visiting.add(agent_id) + for child_id in records[agent_id].managed_agents: + visit(child_id) + visiting.remove(agent_id) + visited.add(agent_id) + + for record_id in records: + visit(record_id) + + async def import_agent_impl( agent_info: ExportAndImportDataFormat, authorization: str = Header(None), @@ -2278,6 +2507,7 @@ async def import_agent_impl( exist for the current tenant. """ user_id, tenant_id, _ = get_current_user_info(authorization) + _validate_import_agent_graph(agent_info) agent_id = agent_info.agent_id agent_stack = deque([agent_id]) @@ -2408,6 +2638,9 @@ async def import_agent_by_agent_id( "duty_prompt": import_agent_info.duty_prompt, "constraint_prompt": import_agent_info.constraint_prompt, "few_shots_prompt": import_agent_info.few_shots_prompt, + "runtime_framework": normalize_agent_runtime_framework( + vars(import_agent_info).get("runtime_framework") + ), "enabled": import_agent_info.enabled, "group_ids": user_group_ids}, tenant_id=tenant_id, @@ -2583,6 +2816,9 @@ async def list_all_agent_info_impl(tenant_id: str, user_id: str) -> list[dict]: "is_published": agent.get("current_version_no") is not None, "current_version_no": agent.get("current_version_no"), "is_a2a_server": agent["agent_id"] in a2a_server_agent_ids, + "runtime_framework": normalize_agent_runtime_framework( + agent.get("runtime_framework") + ), }) return simple_agent_list @@ -2747,6 +2983,15 @@ def check_agent_availability( def insert_related_agent_impl(parent_agent_id, child_agent_id, tenant_id): + try: + _validate_related_agent_frameworks( + parent_agent_id=parent_agent_id, + child_agent_ids=[child_agent_id], + tenant_id=tenant_id, + ) + except AppException as exc: + return JSONResponse(status_code=exc.http_status, content=exc.to_dict()) + # search the agent by bfs, check if there is a circular call search_list = deque([child_agent_id]) agent_id_set = set() @@ -3097,6 +3342,12 @@ async def run_agent_stream( user_id=user_id, tenant_id=tenant_id, ) + if not resume and agent_request.agent_id is not None: + _require_agent_runtime_framework_for_run( + agent_id=agent_request.agent_id, + tenant_id=resolved_tenant_id, + version_no=agent_request.version_no or 0, + ) # Auto-create conversation when conversation_id is not provided. # Skip in debug mode: debug runs are ephemeral and must not persist @@ -3476,6 +3727,11 @@ def stop_agent_tasks(conversation_id: int, user_id: str): Stop agent run and preprocess tasks for the specified conversation_id. Matches the behavior of agent_app.agent_stop_api. """ + runtime_stopped = runtime_run_control_registry.request_stop( + conversation_id=conversation_id, + user_id=user_id, + ) + # Stop agent run agent_stopped = agent_run_manager.stop_agent_run(conversation_id, user_id) @@ -3483,8 +3739,10 @@ def stop_agent_tasks(conversation_id: int, user_id: str): preprocess_stopped = preprocess_manager.stop_preprocess_tasks( conversation_id) - if agent_stopped or preprocess_stopped: + if runtime_stopped or agent_stopped or preprocess_stopped: message_parts = [] + if runtime_stopped: + message_parts.append("agent runtime") if agent_stopped: message_parts.append("agent run") if preprocess_stopped: diff --git a/backend/services/agent_version_service.py b/backend/services/agent_version_service.py index fdaf7e89a2..d6c007082f 100644 --- a/backend/services/agent_version_service.py +++ b/backend/services/agent_version_service.py @@ -32,8 +32,12 @@ STATUS_ARCHIVED, ) from database.model_management_db import get_model_by_model_id, get_valid_model_ids +from database.agent_db import search_agent_info_by_agent_id from utils.str_utils import convert_string_to_list +from consts.agent_runtime import normalize_agent_runtime_framework from consts.agent_unavailable_reasons import AgentUnavailableReason +from consts.error_code import ErrorCode +from consts.exceptions import AppException logger = logging.getLogger("agent_version_service") @@ -80,6 +84,47 @@ def publish_version_impl( if not agent_draft: raise ValueError("Agent draft not found") + runtime_framework = normalize_agent_runtime_framework( + agent_draft.get("runtime_framework"), + default=None, + ) + if runtime_framework is None: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_REQUIRED, + "Agent must select a runtime framework before publishing.", + details={"agent_id": agent_id}, + ) + + child_versions: dict[int, int] = {} + for rel in relations_draft: + child_id = rel['selected_agent_id'] + child_version = query_current_version_no(child_id, tenant_id) + if child_version is None: + raise ValueError( + f"Sub-agent {child_id} has no published version; publish the sub-agent first." + ) + child_snapshot = search_agent_info_by_agent_id( + child_id, + tenant_id, + child_version, + ) + child_framework = normalize_agent_runtime_framework( + child_snapshot.get("runtime_framework"), + default=None, + ) + if child_framework != runtime_framework: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_MISMATCH, + "Internal parent and child Agents must use the same runtime framework.", + details={ + "parent_agent_id": agent_id, + "parent_runtime_framework": runtime_framework, + "child_agent_id": child_id, + "child_runtime_framework": child_framework, + }, + ) + child_versions[child_id] = child_version + # Calculate new version number new_version_no = get_next_version_no(agent_id, tenant_id) @@ -106,11 +151,7 @@ def publish_version_impl( # Insert relation snapshots with pinned child agent versions for rel in relations_draft: child_id = rel['selected_agent_id'] - child_version = query_current_version_no(child_id, tenant_id) - if child_version is None: - raise ValueError( - f"Sub-agent {child_id} has no published version; publish the sub-agent first." - ) + child_version = child_versions[child_id] rel_snapshot = rel.copy() rel_snapshot.pop('version_no', None) rel_snapshot['version_no'] = new_version_no @@ -415,6 +456,24 @@ def rollback_version_impl( draft_agent, _, _ = query_agent_draft(agent_id, tenant_id) if not draft_agent: raise ValueError("Agent draft not found") + draft_framework = normalize_agent_runtime_framework( + draft_agent.get("runtime_framework"), + default=None, + ) + target_framework = normalize_agent_runtime_framework( + target_agent.get("runtime_framework"), + default=None, + ) + if draft_framework != target_framework: + raise AppException( + ErrorCode.AGENT_RUNTIME_FRAMEWORK_IMMUTABLE, + "AGENT_RUNTIME_FRAMEWORK_IMMUTABLE", + details={ + "agent_id": agent_id, + "current": draft_framework, + "requested": target_framework, + }, + ) # Get skill snapshots for target version from database import skill_db as skill_db_module @@ -968,6 +1027,9 @@ async def list_published_agents_impl( "current_version_no": agent.get("current_version_no"), "greeting_message": agent.get("greeting_message"), "example_questions": agent.get("example_questions"), + "runtime_framework": normalize_agent_runtime_framework( + agent.get("runtime_framework") + ), }) return simple_agent_list diff --git a/deploy/images/dockerfiles/main/Dockerfile b/deploy/images/dockerfiles/main/Dockerfile index 90987125b7..cd0d5b62ec 100644 --- a/deploy/images/dockerfiles/main/Dockerfile +++ b/deploy/images/dockerfiles/main/Dockerfile @@ -38,6 +38,7 @@ WORKDIR /opt/backend COPY backend/pyproject.toml /opt/backend/pyproject.toml RUN --mount=type=cache,id=nexent-main-uv-${TARGETARCH},target=/root/.cache/uv,sharing=locked \ uv sync --link-mode copy $(test -n "$MIRROR" && echo "-i $MIRROR") +RUN /opt/backend/.venv/bin/python -c "from openjiuwen.core.foundation.tool import LocalFunction, McpServerConfig; from openjiuwen.core.runner import Runner; from openjiuwen.core.single_agent import ReActAgent" RUN mkdir -p "$TIKTOKEN_CACHE_DIR" && \ /opt/backend/.venv/bin/python -c "import tiktoken; tiktoken.get_encoding('cl100k_base')" # Layer 1: install sdk in link mode @@ -58,6 +59,7 @@ COPY --from=builder /opt/tiktoken-cache /opt/tiktoken-cache # Layer 2: copy backend code COPY backend /opt/backend +RUN /opt/backend/.venv/bin/python -c "from services.agent_runtime.providers.openjiuwen_in_process import OpenJiuwenInProcessRuntime, _load_openjiuwen_bindings; _load_openjiuwen_bindings()" COPY VERSION /opt/nexent/VERSION COPY deploy/common/run-sql-migrations.sh deploy/common/start-backend.sh /opt/nexent/scripts/ RUN chmod +x /opt/nexent/scripts/run-sql-migrations.sh /opt/nexent/scripts/start-backend.sh diff --git a/deploy/sql/migrations/v2.3.0_0721_agent_runtime_framework.sql b/deploy/sql/migrations/v2.3.0_0721_agent_runtime_framework.sql new file mode 100644 index 0000000000..014f86f7fa --- /dev/null +++ b/deploy/sql/migrations/v2.3.0_0721_agent_runtime_framework.sql @@ -0,0 +1,36 @@ +-- Persist an immutable execution framework on every Agent version row. + +ALTER TABLE nexent.ag_tenant_agent_t +ADD COLUMN IF NOT EXISTS runtime_framework VARCHAR(20); + +UPDATE nexent.ag_tenant_agent_t +SET runtime_framework = 'smolagents' +WHERE runtime_framework IS NULL; + +ALTER TABLE nexent.ag_tenant_agent_t +DROP CONSTRAINT IF EXISTS ck_ag_tenant_agent_runtime_framework; +ALTER TABLE nexent.ag_tenant_agent_t +ADD CONSTRAINT ck_ag_tenant_agent_runtime_framework +CHECK (runtime_framework IS NULL OR runtime_framework IN ('smolagents', 'openjiuwen')); + +COMMENT ON COLUMN nexent.ag_tenant_agent_t.runtime_framework +IS 'Immutable execution framework: smolagents or openjiuwen'; + +CREATE OR REPLACE FUNCTION nexent.enforce_agent_runtime_framework_immutable() +RETURNS TRIGGER AS $$ +BEGIN + IF OLD.runtime_framework IS NOT NULL + AND NEW.runtime_framework IS DISTINCT FROM OLD.runtime_framework THEN + RAISE EXCEPTION 'AGENT_RUNTIME_FRAMEWORK_IMMUTABLE' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS enforce_agent_runtime_framework_immutable_trigger +ON nexent.ag_tenant_agent_t; +CREATE TRIGGER enforce_agent_runtime_framework_immutable_trigger +BEFORE UPDATE OF runtime_framework ON nexent.ag_tenant_agent_t +FOR EACH ROW +EXECUTE FUNCTION nexent.enforce_agent_runtime_framework_immutable(); diff --git a/deploy/tests/test_common.sh b/deploy/tests/test_common.sh index 90dd0d53c1..35d02a13d9 100755 --- a/deploy/tests/test_common.sh +++ b/deploy/tests/test_common.sh @@ -387,6 +387,9 @@ assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-supabase-kong/templates assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-web/templates/deployment.yaml")" "checksum/nexent-web:" "web deployment should not keep component-named env checksum annotation" assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-openssh/templates/deployment.yaml")" "checksum/nexent-ssh:" "openssh deployment should not keep component-named env checksum annotation" assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-minio/templates/deployment.yaml")" "checksum/nexent-minio" "minio deployment should not keep component-named env checksum annotation" +assert_not_contains "$(cat "$K8S_CHART_DIR/Chart.yaml")" "openjiuwen-runtime" "Helm should not install a second OpenJiuwen runtime" +assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-common/templates/configmap.yaml")" "AGENT_RUNTIME_PROVIDER" "runtime framework selection should come from Agent data" +assert_not_contains "$(cat "$K8S_CHART_DIR/charts/nexent-common/templates/secrets.yaml")" "CAPABILITY_GRANT_SIGNING_KEY" "single-process integration should not create Gateway signing secrets" ENV_CHECKSUM_A="$TMP_DIR/env-checksum-a.env" cat > "$ENV_CHECKSUM_A" <<'ENV' @@ -571,7 +574,12 @@ for compose_file in "$DOCKER_COMPOSE_FILE" "$DOCKER_PROD_COMPOSE_FILE"; do assert_not_contains "$(awk '/^ nexent-mcp:/,/^ nexent-northbound:/' "$compose_file")" "monitoring.env" "docker mcp service should not receive monitoring.env" assert_not_contains "$(awk '/^ nexent-northbound:/,/^ nexent-web:/' "$compose_file")" "monitoring.env" "docker northbound service should not receive monitoring.env" assert_not_contains "$(awk '/^ nexent-data-process:/,/^ redis:/' "$compose_file")" "monitoring.env" "docker data-process service should not receive monitoring.env" + assert_not_contains "$(cat "$compose_file")" "openjiuwen-runtime:" "Compose should not start a second OpenJiuwen runtime" done +assert_not_contains "$(cat "$SCRIPT_DIR/../env/image-source.general.env")" "OPENJIUWEN_RUNTIME_IMAGE" "deployment should not define a second OpenJiuwen image" +assert_not_contains "$(cat "$SCRIPT_DIR/../env/.env.example")" "AGENT_RUNTIME_PROVIDER" "Agent runtime selection should not be deployment-configurable" +assert_contains "$(cat "$SCRIPT_DIR/../images/dockerfiles/main/Dockerfile")" "OpenJiuwenInProcessRuntime" "main image build should verify the in-process runtime" +assert_contains "$(cat "$SCRIPT_DIR/../../backend/pyproject.toml")" 'openjiuwen==0.1.16' "backend should pin the validated OpenJiuwen release" assert_not_contains "$(cat "$DOCKER_DEV_COMPOSE_FILE")" "monitoring.env" "docker dev data-process compose should not receive monitoring.env" assert_contains "$(cat "$SCRIPT_DIR/../docker/compose/docker-compose-monitoring.yml")" 'LANGFUSE_OTLP_AUTH_HEADER: ${LANGFUSE_OTLP_AUTH_HEADER:-}' "docker monitoring compose should pass Langfuse OTLP auth header to the collector" assert_not_contains "$(cat "$SCRIPT_DIR/../docker/compose/docker-compose-monitoring.yml")" "LANGFUSE_OLTP_AUTH_HEADER" "docker monitoring compose should not pass the misspelled Langfuse auth header alias" diff --git a/doc/smolagents-openjiuwen-runtime-acceptance.md b/doc/smolagents-openjiuwen-runtime-acceptance.md new file mode 100644 index 0000000000..ef7ab0e8b9 --- /dev/null +++ b/doc/smolagents-openjiuwen-runtime-acceptance.md @@ -0,0 +1,147 @@ +# Nexent 单服务、Agent 级运行框架验收说明 + +> 日期：2026-07-21 +> 状态：实现与真实环境验收完成；固定使用已有 `openjiuwen==0.1.16`，OpenSpec change 保持未归档 + +## 1. 验收边界 + +通过验收必须同时满足： + +- 仅有 `nexent-runtime:5014` 和现有 `nexent-mcp:5011`，没有 OpenJiuwen Runtime/Gateway 服务或新端口； +- Agent 首次保存时选择框架，之后 UI、API、版本回滚和数据库都无法修改； +- `/agent/run` 只依据 Agent 数据分派，debug/request/tenant 不能覆盖且失败不 fallback； +- Smolagents 路径不导入 OpenJiuwen；OpenJiuwen 首次选择时才在同一进程初始化； +- 本地工具/Knowledge/Memory/Skill/A2A 走 LocalFunction，MCP 直连现有 endpoint； +- 内部父子 Agent同框架、无环，子 Agent作为工具递归运行； +- success/error/cancel/timeout/shutdown 后无请求级资源泄漏。 + +## 2. 当前自动化结果 + +验收结果以实际命令输出为准。最终依赖与代码状态下的结果如下： + +| 范围 | 结果 | 覆盖 | +| --- | ---: | --- | +| 全量 Python runner | 10,955 passed | 272 个 backend/SDK 测试文件，0 failed，总覆盖率 83% | +| OpenJiuwen runtime | 19 passed | 0.1.16 core API、SSL config、stream cleanup、LocalFunction、MCP、A2A、子 Agent、取消、timeout、shutdown | +| Memory/Memory tools | 57 passed | 升级后的 mem0 API 调用与搜索/写入工具回归 | +| 前端 | type-check/build passed；4 passed | TypeScript、生产构建、创建默认/锁定、同框架过滤、保存/复制继承、中英文文案 | +| 部署 | passed | `deploy/tests/test_common.sh`、Compose 与仅 Runtime/MCP 的 Helm render 静态检查 | +| OpenSpec | passed | `openspec validate separate-smolagents-openjiuwen-runtimes --strict` | +| main 镜像 | passed | 两个构建期 import smoke、依赖约束检查和 `pip check` | +| 真实双框架 E2E | passed | 同一 5014 先运行 Smolagents，再懒启动 OpenJiuwen；SSE、final answer、资源注销和消息持久化成功 | +| 真实 MCP endpoint | passed | 从 `nexent-runtime` 直连 `http://nexent-mcp:5011/sse` 并发现已注册工具，无新增 MCP 进程或端口 | + +全量 runner 使用逐文件隔离，避免既有测试模块对 `sys.modules` 的全局 mock 在文件间互相污染。`npm run check-all` +中的 type-check 与 production build 均通过；全仓 lint/format 仍会报告本变更范围外既有的 Agent Repository、A2A 和版本页面 +格式问题，因此未将其误记为本变更通过项。 + +## 3. 数据库与 API 验收 + +### 3.1 数据库 + +执行 fresh init 后继续应用版本迁移，或在已有环境直接应用版本迁移，然后验证： + +```sql +SELECT runtime_framework, count(*) +FROM nexent.ag_tenant_agent_t +GROUP BY runtime_framework; +``` + +预期历史行全部为 `smolagents`。随后分别验证： + +1. 插入 `NULL` 空白 Agent成功； +2. `NULL -> openjiuwen` 成功； +3. `openjiuwen -> openjiuwen` 成功； +4. `openjiuwen -> smolagents` 和 `openjiuwen -> NULL` 触发 `AGENT_RUNTIME_FRAMEWORK_IMMUTABLE`； +5. 非法值违反 check constraint； +6. 草稿和历史版本快照字段一致。 + +### 3.2 API + +- 新客户端显式创建两种框架均成功； +- 旧创建 payload 缺字段得到 Smolagents； +- 已有 OpenJiuwen Agent的旧更新 payload 缺字段仍保留 OpenJiuwen； +- 相同值更新成功，不同值返回 HTTP 409 和 code `030106`； +- 混合内部关系返回 HTTP 409 和 code `030107`； +- NULL 空白 Agent运行/发布前返回 code `030108`； +- 列表、详情、版本、导出和市场快照字段完整； +- 混合框架导入在第一笔写入前失败； +- run/debug payload 即使携带额外同名字段也不能覆盖持久化框架。 + +## 4. 前端验收 + +1. 进入创建页面时 selector 显示 Smolagents； +2. 保存前可切换到 OpenJiuwen，切换后已选内部子 Agent被清空； +3. 首次保存成功后 selector 禁用并显示“创建后不可修改”； +4. 刷新、切换 Agent、版本详情后仍显示正确框架； +5. 复制 Agent后新 Agent继承来源框架且立即锁定； +6. 内部子 Agent候选只显示同框架 Agent，外部 A2A 不过滤； +7. 中英文标签、不可变和冲突提示正确； +8. 非框架字段编辑和保存不受影响。 + +## 5. Runtime 验收 + +### 5.1 Lazy 与并发 + +- 冷启动后只运行 Smolagents，`openjiuwen` 不在已导入模块且 Runner 未启动； +- 首次 OpenJiuwen run 才 import 0.1.16 core API 并启动 Runner； +- 同一进程交替及并发运行两种框架； +- OpenJiuwen 初始化/模型/工具失败只产生 OpenJiuwen failure，不调用 `agent_run()`； +- shutdown 只关闭已初始化 provider。 + +### 5.2 OpenJiuwen 能力 + +- 纯对话与 history； +- 不同模型配置； +- Knowledge 检索与 source 事件； +- Memory 层级、共享/禁用策略及运行后写入； +- Skill sandbox、脚本执行、artifact 捕获与上传； +- 普通本地工具输入 schema 和结果事件； +- 外部 A2A 代理； +- 两层及以上同框架子 Agent，仅根产生外层 final answer； +- stop、timeout、模型错误、工具错误和进程 shutdown 清理。 + +### 5.3 MCP + +- 连接现有 `nexent-mcp:5011`，没有新增监听端口； +- 连接 Agent 已配置的外部 SSE/Streamable HTTP endpoint； +- 一个 server 暴露额外工具时，只绑定 Agent allowlist； +- required server/tool 缺失阻止运行；optional 缺失产生 warning； +- header/token 不出现在 Agent JSON、event、异常、日志或 repr； +- 同 server 并发运行使用不同 request-scoped ID，结束后 server/tool/callback 数量回到基线。 + +## 6. 部署验收 + +运行： + +```bash +bash deploy/tests/test_common.sh +``` + +并检查 Compose/Helm 渲染结果： + +- 不含 `openjiuwen-runtime` service、subchart、profile 或 image 变量； +- 不含 `AGENT_RUNTIME_PROVIDER`、远程 Runtime URL/timeout、Capability Gateway/grant/signing key； +- `nexent-runtime` 仍使用 main 镜像、原启动命令和 5014； +- `nexent-mcp` 仍使用原进程和 5011； +- 实际启动后的监听端口与容器/Pod 数量没有新增 Runtime/MCP 项。 + +## 7. 0.1.16 依赖门禁 + +最终采用以下兼容依赖组合： + +1. `backend/pyproject.toml` 和 `backend/uv.lock` 均固定为 0.1.16； +2. SDK 使用 `openai>=1.108.0`、`mem0ai==1.0.0` 和 `orjson>=3.11.5`，避免 0.1.117 的 OpenAI `<1.100.0` 上限及 3.10.0 的 orjson 冲突； +3. 最终镜像实际解析为 `openjiuwen 0.1.16`、`openai 2.46.0`、`mem0ai 1.0.0`、`orjson 3.11.9`，所有声明约束满足； +4. main 镜像的 core API 与 in-process provider 两个 import smoke 都成功； +5. 最终镜像已完成真实模型、SSE、持久化和现有 MCP endpoint E2E。 + +任何一项失败都不得归档 OpenSpec change，也不得通过修改 Agent框架或 fallback 掩盖故障。 + +## 8. 回滚验收 + +- 应用回滚不修改数据库中的 `runtime_framework`； +- 版本回滚只允许相同框架快照； +- 不存在部署级开关把 OpenJiuwen Agent改成 Smolagents； +- 如需框架转换，复制业务配置并创建新的目标框架 Agent； +- 回滚后 Smolagents golden、已有 OpenJiuwen Agent的明确不可用提示和数据完整性均可验证。 diff --git a/doc/smolagents-openjiuwen-runtime-integration-design.md b/doc/smolagents-openjiuwen-runtime-integration-design.md new file mode 100644 index 0000000000..d6ea6762e0 --- /dev/null +++ b/doc/smolagents-openjiuwen-runtime-integration-design.md @@ -0,0 +1,180 @@ +# Nexent 单服务、Agent 级运行框架设计 + +> 状态：实现与真实环境验收完成（固定使用已有 `openjiuwen==0.1.16`） +> 日期：2026-07-21 +> 涉及仓库：`nexent`、`/Users/hsc/Applications/opensource/agent-core` + +## 1. 结论 + +Nexent 只保留现有 Runtime 和 MCP 服务： + +| 服务 | 端口 | 职责 | +| --- | ---: | --- | +| `nexent-runtime` | 5014 | 统一 `/agent/run`、SSE、持久化，并在进程内执行 Smolagents 或 OpenJiuwen | +| `nexent-mcp` | 5011 | 继续提供现有 MCP endpoint | + +不新增 OpenJiuwen service、进程、镜像、端口、Capability Gateway 或 Runtime 回环 HTTP。Agent 在首次保存时选择 +`smolagents` 或 `openjiuwen`，框架随后永久不可修改。运行失败形成所选框架的失败事件，不自动切换框架。 + +```mermaid +flowchart LR + Client["页面 / API"] --> Run["nexent-runtime:5014 /agent/run"] + Run --> Read["读取 Agent.runtime_framework"] + Read -->|smolagents| Smol["现有 agent_run()"] + Read -->|openjiuwen| Jiuwen["进程内 ReActAgent"] + Jiuwen --> Local["LocalFunction：工具 / 知识 / 记忆 / Skill / A2A"] + Jiuwen --> MCP["McpServerConfig：现有 MCP endpoint"] + MCP --> NexentMCP["nexent-mcp:5011"] + MCP --> ExternalMCP["已配置的外部 MCP"] + Smol --> Sink["同一 Event Sink / SSE / DB"] + Jiuwen --> Sink +``` + +OpenJiuwen standalone 仍在 agent-core 中按原生方式独立运行，不依赖 Nexent。 + +## 2. 数据模型与不可变规则 + +`nexent.ag_tenant_agent_t` 的每个草稿和版本快照增加 nullable `runtime_framework`： + +- 合法非空值只有 `smolagents`、`openjiuwen`。 +- 迁移将历史行回填为 `smolagents`。 +- 新预创建的空白子 Agent 可以暂时为 `NULL`；首次保存时必须赋值。 +- 数据库 trigger 允许 `NULL -> 合法值` 和相同值幂等更新，拒绝任何非空值变化及非空值改回 `NULL`。 +- 服务层在数据库写入前返回 `409 AGENT_RUNTIME_FRAMEWORK_IMMUTABLE`，数据库 trigger 是最后防线。 +- 旧创建 API 未传字段时默认 Smolagents；旧更新 API 未传字段时保留已有框架。 + +字段贯穿 Agent 创建/更新、详情、列表、版本、发布/回滚、复制、导入导出和市场安装。旧导入格式缺字段时按 +Smolagents 处理。版本回滚不得改变草稿的框架。 + +`/agent/run`、debug 参数和租户配置都没有框架覆盖字段，运行时只读取装配后的 `AgentRunInfo.runtime_framework`。 + +## 3. 页面行为 + +创建页面默认选中 Smolagents，也允许选择 OpenJiuwen。第一次保存成功后： + +- selector 永久禁用并显示“创建后不可修改”； +- store、baseline、脏检查和保存 payload 都保留 `runtime_framework`； +- 切换尚未保存的新 Agent 框架时清空已选内部子 Agent； +- 内部子 Agent选择器只显示相同框架的候选； +- 复制继承来源框架；后端仍负责最终不可变与关系校验。 + +数据库值为 `NULL` 的空白 Agent在页面上显示默认 Smolagents，但保持未锁定，首次保存才完成赋值。 + +## 4. 父子 Agent约束 + +内部父子 Agent必须使用相同框架。校验覆盖： + +1. 页面候选过滤； +2. Agent 批量保存关系； +3. 独立关系 API； +4. 导入图预检； +5. 版本发布时的子版本固定； +6. 运行前递归装配。 + +混合框架关系返回 `409 AGENT_RUNTIME_FRAMEWORK_MISMATCH`。导入在创建任何 Agent 前完成整图框架和环检测。 +运行装配再次验证同框架与无环，防止绕过页面或服务层的数据进入执行阶段。 + +外部 A2A Agent不参与该约束。它继续按协议代理，在 OpenJiuwen 中包装成请求级 `LocalFunction(task)`。 + +## 5. 单进程 Runtime 分派 + +`backend/services/agent_runtime/registry.py` 同时声明两个 factory path，但不提前导入 provider： + +- `SmolagentsRuntime` 委托现有 `agent_run(AgentRunInfo)`； +- `OpenJiuwenInProcessRuntime` 仅在第一个 OpenJiuwen Agent运行时导入 OpenJiuwen、启动全局 `Runner`。 + +因此，只运行 Smolagents 时不会导入 `openjiuwen`。OpenJiuwen 初始化后，两种 provider 可在同一个 +`nexent-runtime` 进程中并发服务不同 Agent。 + +统一的 `AgentRuntimeExecution` 携带 run ID、`AgentRunInfo`、conversation、user、tenant 和 version。两个 provider +都输出既有 Nexent chunk，继续经过同一 message unit、SSE、resume 和持久化逻辑。 + +OpenJiuwen 初始化、装配、模型或工具错误都会产生显式 error event 并终止运行；registry 不包含 fallback 分支。 + +## 6. OpenJiuwen 运行与资源生命周期 + +每次根运行建立独立 active-run 记录和取消信号。每个 Agent节点建立独立 `_NodeScope`，拥有： + +- `ReActAgent`、session、context 和 callback； +- 本节点的 Nexent 工具实例； +- 请求级 LocalFunction 与 MCP server/tool； +- request/run/agent 作用域 ID。 + +stop endpoint 仍按 conversation/user 工作，内部解析为 run ID，并取消对应 OpenJiuwen producer、子 Agent和工具调用。 +Nexent 的 `stop_event` 同时置位，使现有工具、A2A client 和 Smolagents 保持原取消语义。 + +结束、错误和取消均按子到父的调用栈释放 callback、ability、context、MCP server、工具实例和连接。应用退出只关闭 +已经初始化的 provider；OpenJiuwen 未运行过时不会 import 或调用 `Runner.stop()`。 + +## 7. 本地能力复用 + +Knowledge、Memory、Skill 和普通工具不经过 MCP。装配阶段继续使用现有权限、memory policy、skill sandbox、模型和 +artifact 配置生成 `ToolConfig`；OpenJiuwen 运行时通过 `NexentAgent.create_tool()` 创建请求级实例，再包装成 +`ToolCard + LocalFunction`。 + +包装器保留原输入 schema，调用现有 `forward()`/callable，并把 `MessageObserver` 中的检索、工具日志、Skill 文件等 +既有事件送入共享 emitter。Skill artifact 仍由统一流处理链完成路径校验和上传，不引入 OpenJiuwen 自有数据源。 + +外部 A2A 使用现有 `ExternalA2AAgentWrapper` 和认证配置，同样包装为 LocalFunction。 + +## 8. MCP 复用与隔离 + +Agent 装配只查询一次现有 MCP 数据，并同时生成： + +- `mcp_host`：供 Smolagents 现有路径使用； +- `mcp_bindings`：供 OpenJiuwen 使用的结构化绑定。 + +每个 `MCPBinding` 包含 server ID/name、现有 URL、transport、请求级 header、Agent 选择的工具 allowlist、required tool +列表和 endpoint 可用状态。认证 header 标记为不参与 Pydantic 序列化与 repr，不进入 Agent 数据、事件或日志。 + +OpenJiuwen 为每个 Agent节点生成唯一 `McpServerConfig.server_id`，直接连接 `nexent-mcp:5011` 或已有外部 endpoint： + +- 不启动 MCP server； +- 不创建新端口； +- 不将同 server 的未选工具加入 ability； +- required server/工具缺失时阻止运行； +- optional server/工具失败时发送受控 warning； +- success、error、cancel、timeout 后移除 server、tool 和 callback resources。 + +## 9. 同框架子 Agent + +运行前从 `AgentConfig` 构建递归 `OpenJiuwenRunSpec`。每个节点记录 Agent ID/name、模型、prompt、工具/MCP binding、 +父 ID 和 depth。 + +父 Agent把每个子 Agent暴露为 `LocalFunction(task)`。调用时创建子节点独立 session/context/scope；子 Agent最终答案 +作为父 Agent的 tool result。所有节点共享 sequence emitter 和根取消信号，但只有根节点产生外层 `final_answer`。 +子节点事件附带 `agent_id`、`agent_name`、`parent_agent_id`、`depth` 和全局 sequence。 + +该设计复用现有 managed-agent 语义，不启动 AgentTeam 服务或任何额外进程。 + +## 10. 镜像与部署 + +main 镜像固定安装已有 `openjiuwen==0.1.16`。Nexent provider 首次运行时直接懒加载其公开的 `Runner`、`ReActAgent`、 +`LocalFunction` 和 `McpServerConfig` 等 core API，不要求 `openjiuwen.extensions.nexent`。镜像构建在依赖层检查这些 core API， +复制 backend 后再 import `OpenJiuwenInProcessRuntime` 并调用同一 bindings loader，使缺包或 API 不兼容在构建期失败。 + +0.1.16 声明 `openai>=1.108.0`，因此 Nexent SDK 使用 `mem0ai==1.0.0` 取代限制 `openai<1.100.0` 的 0.1.117；同时使用 +`orjson>=3.11.5` 满足 OpenJiuwen 依赖树中的 langgraph-sdk 与 pymilvus。最终 main 镜像的解析结果由构建和依赖检查门禁验证， +不得在 SDK 安装阶段静默降级成声明约束不兼容的版本。 + +Compose 和 Helm 不定义 OpenJiuwen Runtime service/subchart/profile、独立 image、Runtime URL、Capability Gateway、grant +或 signing key。`nexent-runtime` 的命令、镜像和 5014 端口不变，`nexent-mcp` 的进程和 5011 端口不变。 + +## 11. 发布与回滚 + +发布顺序： + +1. 固定 Nexent 依赖和 lock 为已有 `openjiuwen==0.1.16`； +2. 构建 main 镜像并验证 core API 与 in-process provider import smoke； +3. 运行 Compose/Helm 静态与启动 smoke； +4. 先验证 Smolagents golden，再创建新的 OpenJiuwen Agent灰度； +5. 验证并发、stop、MCP header 隔离和资源基线。 + +回滚只回滚应用版本或停止创建新的 OpenJiuwen Agent。框架是 Agent 的不可变业务数据，不能通过部署配置或版本回滚把 +已有 OpenJiuwen Agent改成 Smolagents；需要转换时必须新建 Agent。OpenJiuwen 运行失败也不会自动回退。 + +## 12. 验收结果 + +0.1.16 的公开 core API 已在最终 `nexent-runtime` 容器中完成真实模型运行、Runner 生命周期、SSE、持久化和资源注销验证； +四个 main-image 服务使用同一个 `nexent/nexent:latest` 重新创建，现有 `nexent-mcp:5011/sse` 可从 Runtime 直接发现工具。 +Knowledge、Memory、Skill、A2A、递归子 Agent、MCP allowlist/required/optional/header 隔离和 cleanup 由全量自动化覆盖。 diff --git a/frontend/app/[locale]/agents/components/AgentSelectorHeader.tsx b/frontend/app/[locale]/agents/components/AgentSelectorHeader.tsx index 2dcbe0290d..b9c9098320 100644 --- a/frontend/app/[locale]/agents/components/AgentSelectorHeader.tsx +++ b/frontend/app/[locale]/agents/components/AgentSelectorHeader.tsx @@ -299,6 +299,7 @@ export default function AgentSelectorHeader({ business_logic_model_id: detail.business_logic_model_id ?? undefined, enabled_tool_ids: enabledToolIds, related_agent_ids: subAgentIds, + runtime_framework: detail.runtime_framework || "smolagents", }); if (!createResult.success || !createResult.data?.agent_id) { diff --git a/frontend/app/[locale]/agents/components/agentConfig/CollaborativeAgent.tsx b/frontend/app/[locale]/agents/components/agentConfig/CollaborativeAgent.tsx index d3090b369f..47ce6a1069 100644 --- a/frontend/app/[locale]/agents/components/agentConfig/CollaborativeAgent.tsx +++ b/frontend/app/[locale]/agents/components/agentConfig/CollaborativeAgent.tsx @@ -39,14 +39,18 @@ export default function CollaborativeAgent() { // Related internal agent IDs const relatedAgentIds = Array.isArray(editedAgent?.sub_agent_id_list) ? editedAgent.sub_agent_id_list : []; + const runtimeFramework = editedAgent?.runtime_framework || "smolagents"; + const sameFrameworkInternalAgents = (Array.isArray(internalAgents) ? internalAgents : []).filter( + (agent: Agent) => (agent.runtime_framework || "smolagents") === runtimeFramework + ); // Related internal agents (from published list) - const relatedInternalAgents = (Array.isArray(internalAgents) ? internalAgents : []).filter( + const relatedInternalAgents = sameFrameworkInternalAgents.filter( (agent: Agent) => relatedAgentIds.includes(Number(agent.id)) ); // Available internal agents (exclude already related ones and current agent) - const availableInternalAgents = (Array.isArray(internalAgents) ? internalAgents : []).filter( + const availableInternalAgents = sameFrameworkInternalAgents.filter( (agent: Agent) => !relatedAgentIds.includes(Number(agent.id)) && Number(agent.id) !== currentAgentId ); diff --git a/frontend/app/[locale]/agents/components/agentInfo/AgentGenerateDetail.tsx b/frontend/app/[locale]/agents/components/agentInfo/AgentGenerateDetail.tsx index 17d19ffab5..7e19524440 100644 --- a/frontend/app/[locale]/agents/components/agentInfo/AgentGenerateDetail.tsx +++ b/frontend/app/[locale]/agents/components/agentInfo/AgentGenerateDetail.tsx @@ -68,7 +68,11 @@ export default function AgentGenerateDetail({}) { const currentAgentId = useAgentConfigStore((state) => state.currentAgentId); const forceRefreshKey = useAgentConfigStore((state) => state.forceRefreshKey); const isReadOnly = useAgentConfigStore((state) => state.isReadOnly()); + const isRuntimeFrameworkLocked = useAgentConfigStore( + (state) => state.isRuntimeFrameworkLocked + ); const updateAgentConfig = useAgentConfigStore((state) => state.updateAgentConfig); + const updateSubAgentIds = useAgentConfigStore((state) => state.updateSubAgentIds); const isGenerating = useAgentConfigStore((state) => state.isGenerating); // Determine if form should be editable (based on isReadOnly only, isGenerating handled separately) @@ -238,6 +242,7 @@ export default function AgentGenerateDetail({}) { businessLogicModelId: editedAgent.business_logic_model_id, promptTemplateId: editedAgent.prompt_template_id, promptTemplateName: editedAgent.prompt_template_name || "system_default", + runtimeFramework: editedAgent.runtime_framework || "smolagents", }; queueMicrotask(() => { form.setFieldsValue(initialAgentInfo); @@ -962,6 +967,88 @@ export default function AgentGenerateDetail({}) { /> + + + + { + const nextGroupIds = normalizeNumberArray(value || []); + const currentGroupIds = normalizeNumberArray( + editedAgent.group_ids || [] + ); + if ( + JSON.stringify(nextGroupIds) === + JSON.stringify(currentGroupIds) + ) { + return; + } + updateAgentConfig({ group_ids: nextGroupIds }); + }} + /> + + + + +