opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 02:28:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(agent): import workflow agents as inline (#39135)

Browse Source
This commit is contained in:
zyssyz123 2026-07-16 16:28:34 +08:00 committed by GitHub
parent bfffcb7d0f
commit cd98193234
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 174 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
api/services/agent/dsl_service.py
View File

@ -11,17 +11,14 @@ from __future__ import annotations
import copy
import json
import logging
from collections.abc import Mapping
from typing import Any, cast
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import event, func, select
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from constants.model_template import default_app_templates
from core.workflow.nodes.agent_v2.validators import WorkflowAgentNodeValidator
from events.app_event import app_was_created
from graphon.enums import BuiltinNodeTypes
from models import Account
from models.agent import (
@ -41,7 +38,7 @@ from models.agent import (
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import AgentSoulConfig, WorkflowNodeJobConfig
from models.model import App, AppMode, AppModelConfig, IconType
from models.model import App, AppModelConfig
from models.workflow import Workflow
from services.agent.agent_soul_state import agent_soul_has_model
from services.agent.dsl_entities import (
@ -57,8 +54,6 @@ from services.agent.roster_service import AgentRosterService
from services.entities.dsl_entities import DslImportWarning
from services.plugin.dependencies_analysis import DependenciesAnalysisService
logger = logging.getLogger(__name__)
class AgentPackageImportResult(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
@ -249,7 +244,7 @@ class AgentDslService:
raw_packages: Mapping[str, Any],
account: Account,
) -> tuple[dict[str, Any], list[DslImportWarning]]:
"""Materialize packages and bindings for a Workflow or Snippet draft."""
"""Materialize every packaged Agent as a node-owned inline Agent."""
graph = copy.deepcopy(dict(portable_graph))
packages = {key: AgentPackage.model_validate(value) for key, value in raw_packages.items()}
@ -264,7 +259,6 @@ class AgentDslService:
for binding in previous_bindings:
self.session.delete(binding)
self.session.flush()
imported_roster: dict[str, AgentPackageImportResult] = {}
warnings: list[DslImportWarning] = []
for node_id, raw_node_data in WorkflowAgentNodeValidator.iter_agent_v2_nodes(graph):
@ -280,28 +274,17 @@ class AgentDslService:
raise ValueError(f"Workflow Agent node {node_id} references unknown package {package_ref!r}.")
try:
binding_type = WorkflowAgentBindingType(str(raw_binding.get("binding_type")))
WorkflowAgentBindingType(str(raw_binding.get("binding_type")))
except ValueError as exc:
raise ValueError(f"Workflow Agent node {node_id} has an invalid binding type.") from exc
if binding_type == WorkflowAgentBindingType.ROSTER_AGENT:
imported = imported_roster.get(package_ref)
if imported is None:
imported = self._create_imported_roster_agent_app(
tenant_id=workflow.tenant_id,
account=account,
package=package,
package_path=f"agent_packages.{package_ref}",
)
imported_roster[package_ref] = imported
else:
imported = self._create_imported_inline_agent(
workflow=workflow,
node_id=node_id,
account=account,
package=package,
package_path=f"agent_packages.{package_ref}",
)
imported = self._create_imported_inline_agent(
workflow=workflow,
node_id=node_id,
account=account,
package=package,
package_path=f"agent_packages.{package_ref}",
)
node_job = WorkflowNodeJobConfig.model_validate(node_data.get(AGENT_NODE_JOB_DSL_KEY) or {})
self.session.add(
@ -311,7 +294,7 @@ class AgentDslService:
workflow_id=workflow.id,
workflow_version=workflow.version,
node_id=node_id,
binding_type=binding_type,
binding_type=WorkflowAgentBindingType.INLINE_AGENT,
agent_id=imported.agent.id,
current_snapshot_id=imported.snapshot.id,
node_job_config=node_job,
@ -320,7 +303,7 @@ class AgentDslService:
)
)
node_data["agent_binding"] = {
"binding_type": binding_type.value,
"binding_type": WorkflowAgentBindingType.INLINE_AGENT.value,
"agent_id": imported.agent.id,
"current_snapshot_id": imported.snapshot.id,
}
@ -402,43 +385,6 @@ class AgentDslService:
)
return dependencies
def _create_imported_roster_agent_app(
self,
*,
tenant_id: str,
account: Account,
package: AgentPackage,
package_path: str,
) -> AgentPackageImportResult:
metadata = package.metadata
app_template = dict(default_app_templates[AppMode.AGENT]["app"])
app = App(**app_template)
app.name = metadata.name
app.description = metadata.description
app.mode = AppMode.AGENT
app.icon_type = self._app_icon_type(metadata.icon_type)
app.icon = metadata.icon
app.icon_background = metadata.icon_background
app.tenant_id = tenant_id
app.enable_site = True
app.enable_api = True
app.created_by = account.id
app.maintainer = account.id
app.updated_by = account.id
self.session.add(app)
self.session.flush()
app_was_created.send(app, account=account, session=self.session)
self._configure_visible_agent_app_after_commit(
tenant_id=tenant_id,
app_id=app.id,
account_id=account.id,
)
result = self.import_agent_app_package(app=app, account=account, package=package)
result.warnings = [
warning.model_copy(update={"path": f"{package_path}.{warning.path}"}) for warning in result.warnings
]
return result
def _create_imported_inline_agent(
self,
*,
@ -654,28 +600,6 @@ class AgentDslService:
)
return next(candidate for candidate in candidates if candidate not in existing)
def _configure_visible_agent_app_after_commit(self, *, tenant_id: str, app_id: str, account_id: str) -> None:
"""Apply external RBAC and web-app visibility only after the DB transaction commits."""
def configure(_session: Session) -> None:
try:
from services.enterprise import rbac_service as enterprise_rbac_service
from services.enterprise.enterprise_service import EnterpriseService
from services.feature_service import FeatureService
enterprise_rbac_service.try_sync_creator_access_policy_member_bindings(
tenant_id,
account_id,
enterprise_rbac_service.RBACResourceType.APP,
app_id,
)
if FeatureService.get_system_features().webapp_auth.enabled:
EnterpriseService.WebAppAuth.update_app_access_mode(app_id, "private")
except Exception:
logger.exception("Failed to configure imported Agent App %s after commit", app_id)
event.listen(self.session, "after_commit", configure, once=True)
def _require_agent(self, *, tenant_id: str, agent_id: str) -> Agent:
agent = self.session.scalar(select(Agent).where(Agent.tenant_id == tenant_id, Agent.id == agent_id).limit(1))
if agent is None:
@ -702,10 +626,6 @@ class AgentDslService:
def _agent_icon_type(value: str | None) -> AgentIconType | None:
return AgentIconType(value) if value else None
@staticmethod
def _app_icon_type(value: str | None) -> IconType:
return IconType(value) if value else IconType.EMOJI
def is_agent_v2_graph(graph: Mapping[str, Any]) -> bool:
return any(

137
api/tests/test_containers_integration_tests/services/test_app_dsl_service.py
View File

@ -11,7 +11,7 @@ import pytest
import yaml
from faker import Faker
from flask import Flask
from sqlalchemy import select
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from core.trigger.constants import (
@ -22,11 +22,24 @@ from core.trigger.constants import (
from extensions.ext_redis import redis_client
from graphon.enums import BuiltinNodeTypes
from models import Account, App, AppMode
from models.agent import Agent, AgentConfigDraft, AgentConfigDraftType, AgentConfigSnapshot, AgentScope, AgentSource
from models.agent import (
Agent,
AgentConfigDraft,
AgentConfigDraftType,
AgentConfigSnapshot,
AgentScope,
AgentSource,
AgentStatus,
WorkflowAgentBindingType,
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import AgentSoulConfig
from models.model import AppModelConfig, IconType
from models.workflow import Workflow, WorkflowType
from services import app_dsl_service
from services.account_service import AccountService, TenantService
from services.agent.dsl_entities import AGENT_PACKAGE_REF_KEY, make_portable_agent_package
from services.agent.dsl_service import AgentDslService
from services.app_dsl_service import (
CHECK_DEPENDENCIES_REDIS_KEY_PREFIX,
CURRENT_DSL_VERSION,
@ -952,6 +965,126 @@ class TestAppDslService:
assert "model_config" in exported_data
assert "dependencies" in exported_data
def test_workflow_package_import_materializes_all_agent_bindings_as_inline(
self, db_session_with_containers: Session, mock_external_service_dependencies
):
app, account = self._create_test_app_and_account(db_session_with_containers, mock_external_service_dependencies)
app.mode = AppMode.WORKFLOW
workflow = Workflow.new(
tenant_id=app.tenant_id,
app_id=app.id,
type=WorkflowType.WORKFLOW.value,
version=Workflow.VERSION_DRAFT,
graph=json.dumps({"nodes": [], "edges": []}),
features=json.dumps({}),
created_by=account.id,
environment_variables=[],
conversation_variables=[],
rag_pipeline_variables=[],
)
db_session_with_containers.add(workflow)
db_session_with_containers.flush()
source_agent = Agent(
tenant_id=app.tenant_id,
name="Portable Agent",
description="Imported into each node",
role="researcher",
scope=AgentScope.ROSTER,
source=AgentSource.AGENT_APP,
status=AgentStatus.ACTIVE,
created_by=account.id,
updated_by=account.id,
)
package = make_portable_agent_package(source_agent, AgentSoulConfig(config_note="portable"))
graph = {
"nodes": [
{
"id": "roster-node",
"data": {
"type": BuiltinNodeTypes.AGENT,
"version": "2",
"agent_binding": {
"binding_type": WorkflowAgentBindingType.ROSTER_AGENT.value,
AGENT_PACKAGE_REF_KEY: "agent_1",
},
},
},
{
"id": "inline-node",
"data": {
"type": BuiltinNodeTypes.AGENT,
"version": "2",
"agent_binding": {
"binding_type": WorkflowAgentBindingType.INLINE_AGENT.value,
AGENT_PACKAGE_REF_KEY: "agent_1",
},
},
},
],
"edges": [],
}
imported_roster_count_before = db_session_with_containers.scalar(
select(func.count())
.select_from(Agent)
.where(
Agent.tenant_id == app.tenant_id,
Agent.scope == AgentScope.ROSTER,
Agent.source == AgentSource.IMPORTED,
)
)
imported_graph, warnings = AgentDslService(db_session_with_containers).import_workflow_packages(
workflow=workflow,
portable_graph=graph,
raw_packages={"agent_1": package.model_dump(mode="json")},
account=account,
)
db_session_with_containers.commit()
assert warnings == []
graph_bindings = [node["data"]["agent_binding"] for node in imported_graph["nodes"]]
assert all(binding["binding_type"] == WorkflowAgentBindingType.INLINE_AGENT.value for binding in graph_bindings)
assert len({binding["agent_id"] for binding in graph_bindings}) == 2
bindings = db_session_with_containers.scalars(
select(WorkflowAgentNodeBinding).where(
WorkflowAgentNodeBinding.tenant_id == app.tenant_id,
WorkflowAgentNodeBinding.workflow_id == workflow.id,
WorkflowAgentNodeBinding.workflow_version == Workflow.VERSION_DRAFT,
)
).all()
assert len(bindings) == 2
assert all(binding.binding_type == WorkflowAgentBindingType.INLINE_AGENT for binding in bindings)
imported_agents = db_session_with_containers.scalars(
select(Agent).where(Agent.id.in_({binding.agent_id for binding in bindings if binding.agent_id}))
).all()
assert len(imported_agents) == 2
assert all(agent.scope == AgentScope.WORKFLOW_ONLY for agent in imported_agents)
assert all(agent.source == AgentSource.IMPORTED for agent in imported_agents)
assert all(agent.app_id == app.id and agent.workflow_id == workflow.id for agent in imported_agents)
assert {agent.workflow_node_id for agent in imported_agents} == {"roster-node", "inline-node"}
assert all(agent.backing_app_id for agent in imported_agents)
backing_apps = db_session_with_containers.scalars(
select(App).where(App.id.in_({agent.backing_app_id for agent in imported_agents if agent.backing_app_id}))
).all()
assert len(backing_apps) == 2
assert all(backing_app.mode == AppMode.AGENT for backing_app in backing_apps)
assert all(backing_app.enable_site is False and backing_app.enable_api is False for backing_app in backing_apps)
imported_roster_count_after = db_session_with_containers.scalar(
select(func.count())
.select_from(Agent)
.where(
Agent.tenant_id == app.tenant_id,
Agent.scope == AgentScope.ROSTER,
Agent.source == AgentSource.IMPORTED,
)
)
assert imported_roster_count_after == imported_roster_count_before
def test_agent_app_dsl_round_trip_creates_unpublished_imported_agent(
self, db_session_with_containers: Session, mock_external_service_dependencies
):

122
api/tests/unit_tests/services/agent/test_agent_dsl_service.py
View File

@ -1,6 +1,6 @@
import json
from types import SimpleNamespace
from unittest.mock import Mock, call
from unittest.mock import Mock
import pytest
from pydantic import ValidationError
@ -19,7 +19,6 @@ from models.agent import (
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import AgentSoulConfig, WorkflowNodeJobConfig
from models.model import App, IconType
from services.agent.dsl_entities import (
AGENT_NODE_JOB_DSL_KEY,
AGENT_PACKAGE_REF_KEY,
@ -361,7 +360,7 @@ def test_import_agent_app_package_creates_config_and_unpublished_draft(monkeypat
assert session.flush.call_count == 2
def test_import_workflow_packages_replaces_bindings_and_reuses_roster_package() -> None:
def test_import_workflow_packages_materializes_every_package_binding_as_inline() -> None:
package = make_portable_agent_package(_agent(), AgentSoulConfig())
graph = {
"nodes": [
@ -388,18 +387,15 @@ def test_import_workflow_packages_replaces_bindings_and_reuses_roster_package()
session = Mock()
session.scalars.return_value.all.return_value = [old_binding]
service = AgentDslService(session)
roster_result = SimpleNamespace(
agent=SimpleNamespace(id="roster-agent"),
snapshot=SimpleNamespace(id="roster-snapshot"),
warnings=[DslImportWarning(code="roster", path="agent", message="roster warning")],
)
inline_result = SimpleNamespace(
agent=SimpleNamespace(id="inline-agent"),
snapshot=SimpleNamespace(id="inline-snapshot"),
warnings=[DslImportWarning(code="inline", path="agent", message="inline warning")],
)
service._create_imported_roster_agent_app = Mock(return_value=roster_result)
service._create_imported_inline_agent = Mock(return_value=inline_result)
imported_results = [
SimpleNamespace(
agent=SimpleNamespace(id=f"inline-agent-{index}"),
snapshot=SimpleNamespace(id=f"inline-snapshot-{index}"),
warnings=[DslImportWarning(code=f"inline-{index}", path="agent", message="inline warning")],
)
for index in range(1, 4)
]
service._create_imported_inline_agent = Mock(side_effect=imported_results)
workflow = SimpleNamespace(
tenant_id="tenant-1",
app_id="app-1",
@ -416,15 +412,25 @@ def test_import_workflow_packages_replaces_bindings_and_reuses_roster_package()
)
session.delete.assert_called_once_with(old_binding)
service._create_imported_roster_agent_app.assert_called_once()
service._create_imported_inline_agent.assert_called_once()
assert [warning.code for warning in warnings] == ["roster", "roster", "inline"]
assert result["nodes"][0]["data"]["agent_binding"]["agent_id"] == "roster-agent"
assert result["nodes"][2]["data"]["agent_binding"]["agent_id"] == "inline-agent"
assert service._create_imported_inline_agent.call_count == 3
assert [call.kwargs["node_id"] for call in service._create_imported_inline_agent.call_args_list] == [
"roster-1",
"roster-2",
"inline",
]
assert [warning.code for warning in warnings] == ["inline-1", "inline-2", "inline-3"]
bindings = [result["nodes"][index]["data"]["agent_binding"] for index in range(3)]
assert [binding["agent_id"] for binding in bindings] == [
"inline-agent-1",
"inline-agent-2",
"inline-agent-3",
]
assert all(binding["binding_type"] == WorkflowAgentBindingType.INLINE_AGENT.value for binding in bindings)
assert AGENT_NODE_JOB_DSL_KEY not in result["nodes"][0]["data"]
assert json.loads(workflow.graph) == result
added_bindings = [item.args[0] for item in session.add.call_args_list]
assert all(isinstance(binding, WorkflowAgentNodeBinding) for binding in added_bindings)
assert all(binding.binding_type == WorkflowAgentBindingType.INLINE_AGENT for binding in added_bindings)
@pytest.mark.parametrize(
@ -546,35 +552,6 @@ def test_extract_package_dependencies_covers_model_tools_and_knowledge(monkeypat
]
def test_create_imported_roster_agent_app_prefixes_warnings(monkeypatch) -> None:
session = Mock()
service = AgentDslService(session)
service._configure_visible_agent_app_after_commit = Mock()
result = SimpleNamespace(
agent=_agent(),
snapshot=_snapshot(),
warnings=[DslImportWarning(code="setup", path="soul.model", message="setup")],
)
service.import_agent_app_package = Mock(return_value=result)
send = Mock()
monkeypatch.setattr("services.agent.dsl_service.app_was_created.send", send)
imported = service._create_imported_roster_agent_app(
tenant_id="tenant-1",
account=SimpleNamespace(id="account-1"),
package=make_portable_agent_package(_agent(), AgentSoulConfig()),
package_path="agent_packages.agent_1",
)
app = session.add.call_args.args[0]
assert isinstance(app, App)
assert app.name == "Portable Agent"
assert app.enable_site is True
assert app.enable_api is True
send.assert_called_once_with(app, account=SimpleNamespace(id="account-1"), session=session)
assert imported.warnings[0].path == "agent_packages.agent_1.soul.model"
def test_create_imported_inline_agent_uses_import_provenance() -> None:
service = AgentDslService(Mock())
soul = AgentSoulConfig(config_note="inline")
@ -724,51 +701,6 @@ def test_unique_roster_name_uses_first_available_suffix() -> None:
assert result == "Agent import 2"
def test_configure_visible_agent_app_runs_after_commit(monkeypatch) -> None:
session = Mock()
listener = Mock()
monkeypatch.setattr("services.agent.dsl_service.event.listen", listener)
service = AgentDslService(session)
service._configure_visible_agent_app_after_commit(
tenant_id="tenant-1",
app_id="app-1",
account_id="account-1",
)
listener.assert_called_once_with(session, "after_commit", listener.call_args.args[2], once=True)
configure = listener.call_args.args[2]
from services.enterprise import rbac_service
from services.enterprise.enterprise_service import EnterpriseService
from services.feature_service import FeatureService
sync = Mock()
update_access = Mock()
monkeypatch.setattr(rbac_service, "try_sync_creator_access_policy_member_bindings", sync)
monkeypatch.setattr(EnterpriseService.WebAppAuth, "update_app_access_mode", update_access)
monkeypatch.setattr(
FeatureService,
"get_system_features",
Mock(return_value=SimpleNamespace(webapp_auth=SimpleNamespace(enabled=False))),
)
configure(session)
update_access.assert_not_called()
FeatureService.get_system_features.return_value = SimpleNamespace(webapp_auth=SimpleNamespace(enabled=True))
configure(session)
update_access.assert_called_once_with("app-1", "private")
assert sync.call_args_list == [
call("tenant-1", "account-1", rbac_service.RBACResourceType.APP, "app-1"),
call("tenant-1", "account-1", rbac_service.RBACResourceType.APP, "app-1"),
]
monkeypatch.setattr(rbac_service, "try_sync_creator_access_policy_member_bindings", Mock(side_effect=RuntimeError))
logger = Mock()
monkeypatch.setattr("services.agent.dsl_service.logger", logger)
configure(session)
logger.exception.assert_called_once()
def test_require_helpers_and_graph_detection() -> None:
session = Mock()
service = AgentDslService(session)
@ -787,7 +719,5 @@ def test_require_helpers_and_graph_detection() -> None:
assert AgentDslService._agent_icon_type(AgentIconType.EMOJI.value) == AgentIconType.EMOJI
assert AgentDslService._agent_icon_type(None) is None
assert AgentDslService._app_icon_type(IconType.IMAGE.value) == IconType.IMAGE
assert AgentDslService._app_icon_type(None) == IconType.EMOJI
assert is_agent_v2_graph({"nodes": [_agent_node("agent")]}) is True
assert is_agent_v2_graph({"nodes": ["invalid", {"data": {"type": "start"}}]}) is False