mirror of
https://github.com/langgenius/dify.git
synced 2026-09-24 23:22:26 +08:00
feat: add new agent (#36284)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
zyssyz123
co-authored by
autofix-ci[bot]
autofix-ci[bot]
parent
d1417bbe4b
commit
d9e90d0fa0
@@ -0,0 +1,4 @@
|
||||
from .composer_service import AgentComposerService
|
||||
from .roster_service import AgentRosterService
|
||||
|
||||
__all__ = ["AgentComposerService", "AgentRosterService"]
|
||||
@@ -0,0 +1,767 @@
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from extensions.ext_database import db
|
||||
from models.agent import (
|
||||
Agent,
|
||||
AgentConfigRevision,
|
||||
AgentConfigRevisionOperation,
|
||||
AgentConfigSnapshot,
|
||||
AgentKind,
|
||||
AgentScope,
|
||||
AgentSource,
|
||||
AgentStatus,
|
||||
WorkflowAgentBindingType,
|
||||
WorkflowAgentNodeBinding,
|
||||
)
|
||||
from models.workflow import Workflow
|
||||
from services.agent.composer_validator import ComposerConfigValidator
|
||||
from services.agent.errors import AgentNameConflictError, AgentNotFoundError, AgentVersionNotFoundError
|
||||
from services.entities.agent_entities import (
|
||||
AgentSoulConfig,
|
||||
ComposerCandidatesResponse,
|
||||
ComposerSavePayload,
|
||||
ComposerSaveStrategy,
|
||||
ComposerVariant,
|
||||
WorkflowNodeJobConfig,
|
||||
)
|
||||
|
||||
|
||||
class AgentComposerService:
|
||||
@classmethod
|
||||
def load_workflow_composer(cls, *, tenant_id: str, app_id: str, node_id: str) -> dict[str, Any]:
|
||||
workflow = cls._get_draft_workflow(tenant_id=tenant_id, app_id=app_id)
|
||||
binding = cls._get_workflow_binding(tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id)
|
||||
if not binding:
|
||||
return cls._empty_workflow_state(app_id=app_id, workflow_id=workflow.id, node_id=node_id)
|
||||
|
||||
agent = cls._get_agent_if_present(tenant_id=tenant_id, agent_id=binding.agent_id)
|
||||
version = cls._get_version_if_present(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id if agent else None,
|
||||
version_id=binding.current_snapshot_id,
|
||||
)
|
||||
return cls._serialize_workflow_state(binding=binding, agent=agent, version=version)
|
||||
|
||||
@classmethod
|
||||
def save_workflow_composer(
|
||||
cls, *, tenant_id: str, app_id: str, node_id: str, account_id: str, payload: ComposerSavePayload
|
||||
) -> dict[str, Any]:
|
||||
if payload.variant != ComposerVariant.WORKFLOW:
|
||||
raise ValueError("Workflow composer endpoint only accepts workflow variant")
|
||||
|
||||
ComposerConfigValidator.validate_save_payload(payload)
|
||||
workflow = cls._get_draft_workflow(tenant_id=tenant_id, app_id=app_id)
|
||||
binding = cls._get_workflow_binding(tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id)
|
||||
|
||||
match payload.save_strategy:
|
||||
case ComposerSaveStrategy.NODE_JOB_ONLY:
|
||||
binding = cls._save_node_job_only(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow.id,
|
||||
node_id=node_id,
|
||||
account_id=account_id,
|
||||
binding=binding,
|
||||
payload=payload,
|
||||
)
|
||||
case ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION:
|
||||
binding = cls._save_to_current_version(
|
||||
tenant_id=tenant_id, account_id=account_id, binding=binding, payload=payload
|
||||
)
|
||||
case ComposerSaveStrategy.SAVE_AS_NEW_VERSION:
|
||||
binding = cls._save_as_new_version(
|
||||
tenant_id=tenant_id, account_id=account_id, binding=binding, payload=payload
|
||||
)
|
||||
case ComposerSaveStrategy.SAVE_AS_NEW_AGENT:
|
||||
binding = cls._save_as_new_agent(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow.id,
|
||||
node_id=node_id,
|
||||
account_id=account_id,
|
||||
binding=binding,
|
||||
payload=payload,
|
||||
)
|
||||
case ComposerSaveStrategy.SAVE_TO_ROSTER:
|
||||
binding = cls._save_to_roster(
|
||||
tenant_id=tenant_id, account_id=account_id, binding=binding, payload=payload
|
||||
)
|
||||
|
||||
db.session.commit()
|
||||
agent = cls._get_agent_if_present(tenant_id=tenant_id, agent_id=binding.agent_id)
|
||||
version = cls._get_version_if_present(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id if agent else None,
|
||||
version_id=binding.current_snapshot_id,
|
||||
)
|
||||
return cls._serialize_workflow_state(binding=binding, agent=agent, version=version)
|
||||
|
||||
@classmethod
|
||||
def load_agent_app_composer(cls, *, tenant_id: str, app_id: str) -> dict[str, Any]:
|
||||
agent = db.session.scalar(
|
||||
select(Agent)
|
||||
.where(
|
||||
Agent.tenant_id == tenant_id,
|
||||
Agent.app_id == app_id,
|
||||
Agent.scope == AgentScope.ROSTER,
|
||||
Agent.status == AgentStatus.ACTIVE,
|
||||
)
|
||||
.order_by(Agent.created_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
if not agent:
|
||||
raise AgentNotFoundError()
|
||||
version = cls._require_version(
|
||||
tenant_id=tenant_id, agent_id=agent.id, version_id=agent.active_config_snapshot_id
|
||||
)
|
||||
return {
|
||||
"variant": ComposerVariant.AGENT_APP.value,
|
||||
"agent": cls._serialize_agent(agent),
|
||||
"active_config_snapshot": cls._serialize_version(version),
|
||||
"agent_soul": version.config_snapshot_dict,
|
||||
"save_options": [
|
||||
ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value,
|
||||
ComposerSaveStrategy.SAVE_AS_NEW_VERSION.value,
|
||||
],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def save_agent_app_composer(
|
||||
cls, *, tenant_id: str, app_id: str, account_id: str, payload: ComposerSavePayload
|
||||
) -> dict[str, Any]:
|
||||
if payload.variant != ComposerVariant.AGENT_APP:
|
||||
raise ValueError("Agent App composer endpoint only accepts agent_app variant")
|
||||
ComposerConfigValidator.validate_save_payload(payload)
|
||||
if payload.agent_soul is None:
|
||||
raise ValueError("agent_soul is required")
|
||||
|
||||
agent = db.session.scalar(
|
||||
select(Agent)
|
||||
.where(
|
||||
Agent.tenant_id == tenant_id,
|
||||
Agent.app_id == app_id,
|
||||
Agent.scope == AgentScope.ROSTER,
|
||||
Agent.status == AgentStatus.ACTIVE,
|
||||
)
|
||||
.order_by(Agent.created_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
if not agent:
|
||||
agent = Agent(
|
||||
tenant_id=tenant_id,
|
||||
name=payload.new_agent_name or "Untitled Agent",
|
||||
description="",
|
||||
agent_kind=AgentKind.DIFY_AGENT,
|
||||
scope=AgentScope.ROSTER,
|
||||
source=AgentSource.AGENT_APP,
|
||||
app_id=app_id,
|
||||
status=AgentStatus.ACTIVE,
|
||||
created_by=account_id,
|
||||
updated_by=account_id,
|
||||
)
|
||||
db.session.add(agent)
|
||||
try:
|
||||
db.session.flush()
|
||||
except IntegrityError as exc:
|
||||
db.session.rollback()
|
||||
raise AgentNameConflictError() from exc
|
||||
|
||||
if payload.save_strategy == ComposerSaveStrategy.SAVE_AS_NEW_VERSION or not agent.active_config_snapshot_id:
|
||||
version = cls._create_config_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
account_id=account_id,
|
||||
agent_soul=payload.agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_NEW_VERSION,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
else:
|
||||
current_snapshot = cls._require_version(
|
||||
tenant_id=tenant_id, agent_id=agent.id, version_id=agent.active_config_snapshot_id
|
||||
)
|
||||
version = cls._update_current_version(
|
||||
current_snapshot=current_snapshot,
|
||||
account_id=account_id,
|
||||
agent_soul=payload.agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
agent.updated_by = account_id
|
||||
|
||||
db.session.commit()
|
||||
return cls.load_agent_app_composer(tenant_id=tenant_id, app_id=app_id)
|
||||
|
||||
@classmethod
|
||||
def get_workflow_candidates(cls, *, app_id: str) -> dict[str, Any]:
|
||||
response = ComposerCandidatesResponse(
|
||||
variant=ComposerVariant.WORKFLOW,
|
||||
allowed_node_job_candidates={
|
||||
"previous_node_outputs": [],
|
||||
"declare_output_types": ["string", "number", "object", "array", "boolean", "file"],
|
||||
"human_contacts": [],
|
||||
},
|
||||
allowed_soul_candidates={
|
||||
"skills_files": [],
|
||||
"dify_tools": [],
|
||||
"cli_tools": [],
|
||||
"knowledge_datasets": [],
|
||||
"human_contacts": [],
|
||||
},
|
||||
)
|
||||
return response.model_dump(mode="json")
|
||||
|
||||
@classmethod
|
||||
def get_agent_app_candidates(cls, *, app_id: str) -> dict[str, Any]:
|
||||
response = ComposerCandidatesResponse(
|
||||
variant=ComposerVariant.AGENT_APP,
|
||||
allowed_node_job_candidates={},
|
||||
allowed_soul_candidates={
|
||||
"skills_files": [],
|
||||
"dify_tools": [],
|
||||
"cli_tools": [],
|
||||
"knowledge_datasets": [],
|
||||
"human_contacts": [],
|
||||
},
|
||||
)
|
||||
return response.model_dump(mode="json")
|
||||
|
||||
@classmethod
|
||||
def calculate_impact(cls, *, tenant_id: str, current_snapshot_id: str) -> dict[str, Any]:
|
||||
bindings = list(
|
||||
db.session.scalars(
|
||||
select(WorkflowAgentNodeBinding).where(
|
||||
WorkflowAgentNodeBinding.tenant_id == tenant_id,
|
||||
WorkflowAgentNodeBinding.current_snapshot_id == current_snapshot_id,
|
||||
)
|
||||
).all()
|
||||
)
|
||||
return {
|
||||
"current_snapshot_id": current_snapshot_id,
|
||||
"workflow_node_count": len(bindings),
|
||||
"bindings": [
|
||||
{
|
||||
"app_id": binding.app_id,
|
||||
"workflow_id": binding.workflow_id,
|
||||
"node_id": binding.node_id,
|
||||
}
|
||||
for binding in bindings
|
||||
],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _save_node_job_only(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
app_id: str,
|
||||
workflow_id: str,
|
||||
node_id: str,
|
||||
account_id: str,
|
||||
binding: WorkflowAgentNodeBinding | None,
|
||||
payload: ComposerSavePayload,
|
||||
) -> WorkflowAgentNodeBinding:
|
||||
node_job = payload.node_job or WorkflowNodeJobConfig()
|
||||
if binding:
|
||||
binding.node_job_config = node_job
|
||||
binding.updated_by = account_id
|
||||
return binding
|
||||
|
||||
agent_soul = payload.agent_soul or AgentSoulConfig()
|
||||
agent = cls._create_workflow_only_agent(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow_id,
|
||||
node_id=node_id,
|
||||
account_id=account_id,
|
||||
agent_soul=agent_soul,
|
||||
)
|
||||
binding = WorkflowAgentNodeBinding(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow_id,
|
||||
node_id=node_id,
|
||||
binding_type=WorkflowAgentBindingType.INLINE_AGENT,
|
||||
agent_id=agent.id,
|
||||
current_snapshot_id=agent.active_config_snapshot_id,
|
||||
node_job_config=node_job,
|
||||
created_by=account_id,
|
||||
updated_by=account_id,
|
||||
)
|
||||
db.session.add(binding)
|
||||
db.session.flush()
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _save_to_current_version(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
account_id: str,
|
||||
binding: WorkflowAgentNodeBinding | None,
|
||||
payload: ComposerSavePayload,
|
||||
) -> WorkflowAgentNodeBinding:
|
||||
binding = cls._require_binding(binding)
|
||||
if payload.agent_soul is None:
|
||||
raise ValueError("agent_soul is required")
|
||||
current_snapshot = cls._require_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=binding.agent_id,
|
||||
version_id=binding.current_snapshot_id,
|
||||
)
|
||||
version = cls._update_current_version(
|
||||
current_snapshot=current_snapshot,
|
||||
account_id=account_id,
|
||||
agent_soul=payload.agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
agent = cls._require_agent(tenant_id=tenant_id, agent_id=binding.agent_id)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
agent.updated_by = account_id
|
||||
binding.current_snapshot_id = version.id
|
||||
if payload.node_job is not None:
|
||||
binding.node_job_config = payload.node_job
|
||||
binding.updated_by = account_id
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _save_as_new_version(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
account_id: str,
|
||||
binding: WorkflowAgentNodeBinding | None,
|
||||
payload: ComposerSavePayload,
|
||||
) -> WorkflowAgentNodeBinding:
|
||||
binding = cls._require_binding(binding)
|
||||
if not binding.agent_id or payload.agent_soul is None:
|
||||
raise ValueError("agent_id and agent_soul are required")
|
||||
version = cls._create_config_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=binding.agent_id,
|
||||
account_id=account_id,
|
||||
agent_soul=payload.agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_NEW_VERSION,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
agent = cls._require_agent(tenant_id=tenant_id, agent_id=binding.agent_id)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
agent.updated_by = account_id
|
||||
binding.current_snapshot_id = version.id
|
||||
binding.updated_by = account_id
|
||||
if payload.node_job is not None:
|
||||
binding.node_job_config = payload.node_job
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _save_as_new_agent(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
app_id: str,
|
||||
workflow_id: str,
|
||||
node_id: str,
|
||||
account_id: str,
|
||||
binding: WorkflowAgentNodeBinding | None,
|
||||
payload: ComposerSavePayload,
|
||||
) -> WorkflowAgentNodeBinding:
|
||||
if payload.agent_soul is None:
|
||||
raise ValueError("agent_soul is required")
|
||||
agent_name = payload.new_agent_name or "Untitled Agent"
|
||||
agent = cls._create_roster_agent_for_composer(
|
||||
tenant_id=tenant_id,
|
||||
account_id=account_id,
|
||||
name=agent_name,
|
||||
agent_soul=payload.agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_NEW_AGENT,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
node_job = payload.node_job or WorkflowNodeJobConfig()
|
||||
if not binding:
|
||||
binding = WorkflowAgentNodeBinding(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow_id,
|
||||
node_id=node_id,
|
||||
created_by=account_id,
|
||||
)
|
||||
db.session.add(binding)
|
||||
binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT
|
||||
binding.agent_id = agent.id
|
||||
binding.current_snapshot_id = agent.active_config_snapshot_id
|
||||
binding.node_job_config = node_job
|
||||
binding.updated_by = account_id
|
||||
db.session.flush()
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _save_to_roster(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
account_id: str,
|
||||
binding: WorkflowAgentNodeBinding | None,
|
||||
payload: ComposerSavePayload,
|
||||
) -> WorkflowAgentNodeBinding:
|
||||
binding = cls._require_binding(binding)
|
||||
source_agent = cls._require_agent(tenant_id=tenant_id, agent_id=binding.agent_id)
|
||||
source_version = cls._require_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=source_agent.id,
|
||||
version_id=binding.current_snapshot_id,
|
||||
)
|
||||
agent_soul = payload.agent_soul or AgentSoulConfig.model_validate(source_version.config_snapshot_dict)
|
||||
agent_name = payload.new_agent_name or source_agent.name
|
||||
roster_agent = cls._create_roster_agent_for_composer(
|
||||
tenant_id=tenant_id,
|
||||
account_id=account_id,
|
||||
name=agent_name,
|
||||
agent_soul=agent_soul,
|
||||
operation=AgentConfigRevisionOperation.SAVE_TO_ROSTER,
|
||||
version_note=payload.version_note,
|
||||
)
|
||||
binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT
|
||||
binding.agent_id = roster_agent.id
|
||||
binding.current_snapshot_id = roster_agent.active_config_snapshot_id
|
||||
binding.updated_by = account_id
|
||||
if payload.node_job is not None:
|
||||
binding.node_job_config = payload.node_job
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _create_workflow_only_agent(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
app_id: str,
|
||||
workflow_id: str,
|
||||
node_id: str,
|
||||
account_id: str,
|
||||
agent_soul: AgentSoulConfig,
|
||||
) -> Agent:
|
||||
agent = Agent(
|
||||
tenant_id=tenant_id,
|
||||
name=f"Workflow Agent {node_id}",
|
||||
description="",
|
||||
agent_kind=AgentKind.DIFY_AGENT,
|
||||
scope=AgentScope.WORKFLOW_ONLY,
|
||||
source=AgentSource.WORKFLOW,
|
||||
app_id=app_id,
|
||||
workflow_id=workflow_id,
|
||||
workflow_node_id=node_id,
|
||||
status=AgentStatus.ACTIVE,
|
||||
created_by=account_id,
|
||||
updated_by=account_id,
|
||||
)
|
||||
db.session.add(agent)
|
||||
db.session.flush()
|
||||
version = cls._create_config_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
account_id=account_id,
|
||||
agent_soul=agent_soul,
|
||||
operation=AgentConfigRevisionOperation.CREATE_VERSION,
|
||||
version_note=None,
|
||||
)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
return agent
|
||||
|
||||
@classmethod
|
||||
def _create_roster_agent_for_composer(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
account_id: str,
|
||||
name: str,
|
||||
agent_soul: AgentSoulConfig,
|
||||
operation: AgentConfigRevisionOperation,
|
||||
version_note: str | None,
|
||||
) -> Agent:
|
||||
agent = Agent(
|
||||
tenant_id=tenant_id,
|
||||
name=name,
|
||||
description="",
|
||||
agent_kind=AgentKind.DIFY_AGENT,
|
||||
scope=AgentScope.ROSTER,
|
||||
source=AgentSource.WORKFLOW,
|
||||
status=AgentStatus.ACTIVE,
|
||||
created_by=account_id,
|
||||
updated_by=account_id,
|
||||
)
|
||||
db.session.add(agent)
|
||||
try:
|
||||
db.session.flush()
|
||||
except IntegrityError as exc:
|
||||
db.session.rollback()
|
||||
raise AgentNameConflictError() from exc
|
||||
version = cls._create_config_version(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
account_id=account_id,
|
||||
agent_soul=agent_soul,
|
||||
operation=operation,
|
||||
version_note=version_note,
|
||||
)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
return agent
|
||||
|
||||
@classmethod
|
||||
def _create_config_version(
|
||||
cls,
|
||||
*,
|
||||
tenant_id: str,
|
||||
agent_id: str,
|
||||
account_id: str,
|
||||
agent_soul: AgentSoulConfig,
|
||||
operation: AgentConfigRevisionOperation,
|
||||
version_note: str | None,
|
||||
previous_snapshot_id: str | None = None,
|
||||
) -> AgentConfigSnapshot:
|
||||
next_version = (
|
||||
db.session.scalar(
|
||||
select(func.max(AgentConfigSnapshot.version)).where(
|
||||
AgentConfigSnapshot.tenant_id == tenant_id,
|
||||
AgentConfigSnapshot.agent_id == agent_id,
|
||||
)
|
||||
)
|
||||
or 0
|
||||
) + 1
|
||||
version = AgentConfigSnapshot(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
version=next_version,
|
||||
config_snapshot=agent_soul,
|
||||
version_note=version_note,
|
||||
created_by=account_id,
|
||||
)
|
||||
db.session.add(version)
|
||||
db.session.flush()
|
||||
revision = AgentConfigRevision(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
previous_snapshot_id=previous_snapshot_id,
|
||||
current_snapshot_id=version.id,
|
||||
revision=cls._next_revision(tenant_id=tenant_id, agent_id=agent_id),
|
||||
operation=operation,
|
||||
version_note=version_note,
|
||||
created_by=account_id,
|
||||
)
|
||||
db.session.add(revision)
|
||||
db.session.flush()
|
||||
return version
|
||||
|
||||
@classmethod
|
||||
def _update_current_version(
|
||||
cls,
|
||||
*,
|
||||
current_snapshot: AgentConfigSnapshot,
|
||||
account_id: str,
|
||||
agent_soul: AgentSoulConfig,
|
||||
operation: AgentConfigRevisionOperation,
|
||||
version_note: str | None,
|
||||
) -> AgentConfigSnapshot:
|
||||
return cls._create_config_version(
|
||||
tenant_id=current_snapshot.tenant_id,
|
||||
agent_id=current_snapshot.agent_id,
|
||||
account_id=account_id,
|
||||
agent_soul=agent_soul,
|
||||
operation=operation,
|
||||
version_note=version_note,
|
||||
previous_snapshot_id=current_snapshot.id,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _next_revision(cls, *, tenant_id: str, agent_id: str) -> int:
|
||||
return (
|
||||
db.session.scalar(
|
||||
select(func.max(AgentConfigRevision.revision)).where(
|
||||
AgentConfigRevision.tenant_id == tenant_id,
|
||||
AgentConfigRevision.agent_id == agent_id,
|
||||
)
|
||||
)
|
||||
or 0
|
||||
) + 1
|
||||
|
||||
@classmethod
|
||||
def _get_draft_workflow(cls, *, tenant_id: str, app_id: str) -> Workflow:
|
||||
workflow = db.session.scalar(
|
||||
select(Workflow)
|
||||
.where(
|
||||
Workflow.tenant_id == tenant_id,
|
||||
Workflow.app_id == app_id,
|
||||
Workflow.version == Workflow.VERSION_DRAFT,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
if not workflow:
|
||||
raise ValueError("Draft workflow not found")
|
||||
return workflow
|
||||
|
||||
@classmethod
|
||||
def _get_workflow_binding(
|
||||
cls, *, tenant_id: str, workflow_id: str, node_id: str
|
||||
) -> WorkflowAgentNodeBinding | None:
|
||||
return db.session.scalar(
|
||||
select(WorkflowAgentNodeBinding)
|
||||
.where(
|
||||
WorkflowAgentNodeBinding.tenant_id == tenant_id,
|
||||
WorkflowAgentNodeBinding.workflow_id == workflow_id,
|
||||
WorkflowAgentNodeBinding.node_id == node_id,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _require_binding(cls, binding: WorkflowAgentNodeBinding | None) -> WorkflowAgentNodeBinding:
|
||||
if not binding:
|
||||
raise ValueError("Workflow agent binding not found")
|
||||
return binding
|
||||
|
||||
@classmethod
|
||||
def _require_agent(cls, *, tenant_id: str, agent_id: str | None) -> Agent:
|
||||
if not agent_id:
|
||||
raise AgentNotFoundError()
|
||||
agent = db.session.scalar(select(Agent).where(Agent.tenant_id == tenant_id, Agent.id == agent_id).limit(1))
|
||||
if not agent:
|
||||
raise AgentNotFoundError()
|
||||
return agent
|
||||
|
||||
@classmethod
|
||||
def _get_agent_if_present(cls, *, tenant_id: str, agent_id: str | None) -> Agent | None:
|
||||
if not agent_id:
|
||||
return None
|
||||
return db.session.scalar(select(Agent).where(Agent.tenant_id == tenant_id, Agent.id == agent_id).limit(1))
|
||||
|
||||
@classmethod
|
||||
def _require_version(cls, *, tenant_id: str, agent_id: str | None, version_id: str | None) -> AgentConfigSnapshot:
|
||||
if not agent_id or not version_id:
|
||||
raise AgentVersionNotFoundError()
|
||||
version = db.session.scalar(
|
||||
select(AgentConfigSnapshot)
|
||||
.where(
|
||||
AgentConfigSnapshot.tenant_id == tenant_id,
|
||||
AgentConfigSnapshot.agent_id == agent_id,
|
||||
AgentConfigSnapshot.id == version_id,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
if not version:
|
||||
raise AgentVersionNotFoundError()
|
||||
return version
|
||||
|
||||
@classmethod
|
||||
def _get_version_if_present(
|
||||
cls, *, tenant_id: str, agent_id: str | None, version_id: str | None
|
||||
) -> AgentConfigSnapshot | None:
|
||||
if not agent_id or not version_id:
|
||||
return None
|
||||
return db.session.scalar(
|
||||
select(AgentConfigSnapshot)
|
||||
.where(
|
||||
AgentConfigSnapshot.tenant_id == tenant_id,
|
||||
AgentConfigSnapshot.agent_id == agent_id,
|
||||
AgentConfigSnapshot.id == version_id,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _empty_workflow_state(cls, *, app_id: str, workflow_id: str, node_id: str) -> dict[str, Any]:
|
||||
return {
|
||||
"variant": ComposerVariant.WORKFLOW.value,
|
||||
"agent": None,
|
||||
"active_config_snapshot": None,
|
||||
"binding": None,
|
||||
"soul_lock": {"locked": False, "can_unlock": False, "reason": "workflow_only_empty"},
|
||||
"agent_soul": AgentSoulConfig().model_dump(mode="json"),
|
||||
"node_job": WorkflowNodeJobConfig().model_dump(mode="json"),
|
||||
"save_options": [ComposerSaveStrategy.NODE_JOB_ONLY.value, ComposerSaveStrategy.SAVE_TO_ROSTER.value],
|
||||
"impact_summary": None,
|
||||
"app_id": app_id,
|
||||
"workflow_id": workflow_id,
|
||||
"node_id": node_id,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_workflow_state(
|
||||
cls,
|
||||
*,
|
||||
binding: WorkflowAgentNodeBinding,
|
||||
agent: Agent | None,
|
||||
version: AgentConfigSnapshot | None,
|
||||
) -> dict[str, Any]:
|
||||
locked = bool(agent and agent.scope == AgentScope.ROSTER)
|
||||
save_options = [ComposerSaveStrategy.NODE_JOB_ONLY.value]
|
||||
if locked:
|
||||
save_options.extend(
|
||||
[
|
||||
ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value,
|
||||
ComposerSaveStrategy.SAVE_AS_NEW_VERSION.value,
|
||||
ComposerSaveStrategy.SAVE_AS_NEW_AGENT.value,
|
||||
]
|
||||
)
|
||||
else:
|
||||
save_options.append(ComposerSaveStrategy.SAVE_TO_ROSTER.value)
|
||||
return {
|
||||
"variant": ComposerVariant.WORKFLOW.value,
|
||||
"agent": cls._serialize_agent(agent) if agent else None,
|
||||
"active_config_snapshot": cls._serialize_version(version),
|
||||
"binding": {
|
||||
"id": binding.id,
|
||||
"binding_type": binding.binding_type.value,
|
||||
"agent_id": binding.agent_id,
|
||||
"current_snapshot_id": binding.current_snapshot_id,
|
||||
"workflow_id": binding.workflow_id,
|
||||
"node_id": binding.node_id,
|
||||
},
|
||||
"soul_lock": {
|
||||
"locked": locked,
|
||||
"can_unlock": locked,
|
||||
"reason": "roster_agent_shared_version" if locked else "workflow_only_agent",
|
||||
},
|
||||
"agent_soul": cls._workflow_agent_soul_config(version.config_snapshot_dict)
|
||||
if version
|
||||
else AgentSoulConfig().model_dump(mode="json"),
|
||||
"node_job": binding.node_job_config_dict,
|
||||
"save_options": save_options,
|
||||
"impact_summary": cls.calculate_impact(
|
||||
tenant_id=binding.tenant_id, current_snapshot_id=binding.current_snapshot_id
|
||||
)
|
||||
if binding.current_snapshot_id
|
||||
else None,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_agent(cls, agent: Agent) -> dict[str, Any]:
|
||||
return {
|
||||
"id": agent.id,
|
||||
"name": agent.name,
|
||||
"description": agent.description,
|
||||
"scope": agent.scope.value,
|
||||
"status": agent.status.value,
|
||||
"active_config_snapshot_id": agent.active_config_snapshot_id,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def _serialize_version(cls, version: AgentConfigSnapshot | None) -> dict[str, Any] | None:
|
||||
if not version:
|
||||
return None
|
||||
return {
|
||||
"id": version.id,
|
||||
"version": version.version,
|
||||
"version_note": version.version_note,
|
||||
"created_by": version.created_by,
|
||||
"created_at": version.created_at.isoformat() if version.created_at else None,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _workflow_agent_soul_config(config_snapshot: dict[str, Any]) -> dict[str, Any]:
|
||||
agent_soul = dict(config_snapshot)
|
||||
agent_soul["app_features"] = {}
|
||||
agent_soul["app_variables"] = []
|
||||
return agent_soul
|
||||
@@ -0,0 +1,71 @@
|
||||
from typing import Any
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from services.agent.errors import AgentSoulLockedError, InvalidComposerConfigError, PlaintextSecretNotAllowedError
|
||||
from services.entities.agent_entities import (
|
||||
AgentSoulConfig,
|
||||
ComposerSavePayload,
|
||||
ComposerVariant,
|
||||
WorkflowNodeJobConfig,
|
||||
)
|
||||
|
||||
_PLAINTEXT_SECRET_KEYS = {
|
||||
"api_key",
|
||||
"apikey",
|
||||
"authorization",
|
||||
"password",
|
||||
"secret",
|
||||
"secret_key",
|
||||
}
|
||||
|
||||
|
||||
class ComposerConfigValidator:
|
||||
@classmethod
|
||||
def validate_save_payload(cls, payload: ComposerSavePayload) -> None:
|
||||
if payload.variant == ComposerVariant.WORKFLOW and payload.soul_lock.locked and payload.agent_soul is not None:
|
||||
raise AgentSoulLockedError()
|
||||
|
||||
if payload.agent_soul is not None:
|
||||
cls.validate_agent_soul(payload.agent_soul)
|
||||
if payload.node_job is not None:
|
||||
cls.validate_node_job(payload.node_job)
|
||||
|
||||
@classmethod
|
||||
def validate_agent_soul(cls, agent_soul: AgentSoulConfig) -> None:
|
||||
cls._reject_plaintext_secrets(agent_soul.model_dump(mode="json"), path="agent_soul")
|
||||
|
||||
@classmethod
|
||||
def validate_node_job(cls, node_job: WorkflowNodeJobConfig) -> None:
|
||||
cls._reject_plaintext_secrets(node_job.model_dump(mode="json"), path="node_job")
|
||||
|
||||
@classmethod
|
||||
def validate_agent_soul_dict(cls, value: dict[str, Any]) -> AgentSoulConfig:
|
||||
try:
|
||||
config = AgentSoulConfig.model_validate(value)
|
||||
except ValidationError as exc:
|
||||
raise InvalidComposerConfigError(str(exc)) from exc
|
||||
cls.validate_agent_soul(config)
|
||||
return config
|
||||
|
||||
@classmethod
|
||||
def validate_node_job_dict(cls, value: dict[str, Any]) -> WorkflowNodeJobConfig:
|
||||
try:
|
||||
config = WorkflowNodeJobConfig.model_validate(value)
|
||||
except ValidationError as exc:
|
||||
raise InvalidComposerConfigError(str(exc)) from exc
|
||||
cls.validate_node_job(config)
|
||||
return config
|
||||
|
||||
@classmethod
|
||||
def _reject_plaintext_secrets(cls, value: Any, *, path: str) -> None:
|
||||
if isinstance(value, dict):
|
||||
for key, nested in value.items():
|
||||
normalized_key = key.lower().replace("-", "_")
|
||||
nested_path = f"{path}.{key}"
|
||||
if normalized_key in _PLAINTEXT_SECRET_KEYS and isinstance(nested, str) and nested:
|
||||
raise PlaintextSecretNotAllowedError(f"Plaintext secret is not allowed at {nested_path}")
|
||||
cls._reject_plaintext_secrets(nested, path=nested_path)
|
||||
elif isinstance(value, list):
|
||||
for index, nested in enumerate(value):
|
||||
cls._reject_plaintext_secrets(nested, path=f"{path}[{index}]")
|
||||
@@ -0,0 +1,29 @@
|
||||
from werkzeug.exceptions import BadRequest, Conflict, NotFound
|
||||
|
||||
|
||||
class AgentNotFoundError(NotFound):
|
||||
description = "Agent not found."
|
||||
|
||||
|
||||
class AgentVersionNotFoundError(NotFound):
|
||||
description = "Agent config version not found."
|
||||
|
||||
|
||||
class AgentNameConflictError(Conflict):
|
||||
description = "Agent name already exists."
|
||||
|
||||
|
||||
class AgentArchivedError(Conflict):
|
||||
description = "Archived agent cannot be modified."
|
||||
|
||||
|
||||
class AgentSoulLockedError(BadRequest):
|
||||
description = "Agent Soul is locked for this workflow node."
|
||||
|
||||
|
||||
class InvalidComposerConfigError(BadRequest):
|
||||
description = "Invalid agent composer config."
|
||||
|
||||
|
||||
class PlaintextSecretNotAllowedError(BadRequest):
|
||||
description = "Plaintext secret values are not allowed in Agent config."
|
||||
@@ -0,0 +1,320 @@
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from libs.datetime_utils import naive_utc_now
|
||||
from models.agent import (
|
||||
Agent,
|
||||
AgentConfigRevision,
|
||||
AgentConfigRevisionOperation,
|
||||
AgentConfigSnapshot,
|
||||
AgentKind,
|
||||
AgentScope,
|
||||
AgentSource,
|
||||
AgentStatus,
|
||||
WorkflowAgentNodeBinding,
|
||||
)
|
||||
from models.workflow import Workflow
|
||||
from services.agent.composer_validator import ComposerConfigValidator
|
||||
from services.agent.errors import (
|
||||
AgentArchivedError,
|
||||
AgentNameConflictError,
|
||||
AgentNotFoundError,
|
||||
AgentVersionNotFoundError,
|
||||
)
|
||||
from services.entities.agent_entities import RosterAgentCreatePayload, RosterAgentUpdatePayload
|
||||
|
||||
|
||||
class AgentRosterService:
|
||||
def __init__(self, session: Any):
|
||||
self._session = session
|
||||
|
||||
@staticmethod
|
||||
def serialize_agent(agent: Agent, active_version: AgentConfigSnapshot | None = None) -> dict[str, Any]:
|
||||
return {
|
||||
"id": agent.id,
|
||||
"name": agent.name,
|
||||
"description": agent.description,
|
||||
"icon_type": agent.icon_type.value if agent.icon_type else None,
|
||||
"icon": agent.icon,
|
||||
"icon_background": agent.icon_background,
|
||||
"agent_kind": agent.agent_kind.value,
|
||||
"scope": agent.scope.value,
|
||||
"source": agent.source.value,
|
||||
"app_id": agent.app_id,
|
||||
"workflow_id": agent.workflow_id,
|
||||
"workflow_node_id": agent.workflow_node_id,
|
||||
"active_config_snapshot_id": agent.active_config_snapshot_id,
|
||||
"active_config_snapshot": AgentRosterService.serialize_version(active_version) if active_version else None,
|
||||
"status": agent.status.value,
|
||||
"created_by": agent.created_by,
|
||||
"updated_by": agent.updated_by,
|
||||
"archived_by": agent.archived_by,
|
||||
"archived_at": agent.archived_at.isoformat() if agent.archived_at else None,
|
||||
"created_at": agent.created_at.isoformat() if agent.created_at else None,
|
||||
"updated_at": agent.updated_at.isoformat() if agent.updated_at else None,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def serialize_version(version: AgentConfigSnapshot | None) -> dict[str, Any] | None:
|
||||
if version is None:
|
||||
return None
|
||||
return {
|
||||
"id": version.id,
|
||||
"agent_id": version.agent_id,
|
||||
"version": version.version,
|
||||
"summary": version.summary,
|
||||
"version_note": version.version_note,
|
||||
"created_by": version.created_by,
|
||||
"created_at": version.created_at.isoformat() if version.created_at else None,
|
||||
}
|
||||
|
||||
def list_roster_agents(
|
||||
self, *, tenant_id: str, page: int = 1, limit: int = 20, keyword: str | None = None
|
||||
) -> dict[str, Any]:
|
||||
stmt = select(Agent).where(
|
||||
Agent.tenant_id == tenant_id,
|
||||
Agent.scope == AgentScope.ROSTER,
|
||||
Agent.status == AgentStatus.ACTIVE,
|
||||
)
|
||||
if keyword:
|
||||
from libs.helper import escape_like_pattern
|
||||
|
||||
escaped_keyword = escape_like_pattern(keyword)
|
||||
stmt = stmt.where(Agent.name.ilike(f"%{escaped_keyword}%", escape="\\"))
|
||||
stmt = stmt.order_by(Agent.updated_at.desc())
|
||||
|
||||
total = self._session.scalar(select(func.count()).select_from(stmt.subquery())) or 0
|
||||
agents = list(self._session.scalars(stmt.offset((page - 1) * limit).limit(limit)).all())
|
||||
versions_by_id = self._load_versions_by_id(
|
||||
[agent.active_config_snapshot_id for agent in agents if agent.active_config_snapshot_id]
|
||||
)
|
||||
|
||||
data = []
|
||||
for agent in agents:
|
||||
active_version = (
|
||||
versions_by_id.get(agent.active_config_snapshot_id) if agent.active_config_snapshot_id else None
|
||||
)
|
||||
data.append(self.serialize_agent(agent, active_version))
|
||||
|
||||
return {
|
||||
"data": data,
|
||||
"page": page,
|
||||
"limit": limit,
|
||||
"total": total,
|
||||
"has_more": page * limit < total,
|
||||
}
|
||||
|
||||
def list_invite_options(
|
||||
self, *, tenant_id: str, page: int = 1, limit: int = 20, keyword: str | None = None, app_id: str | None = None
|
||||
) -> dict[str, Any]:
|
||||
result = self.list_roster_agents(tenant_id=tenant_id, page=page, limit=limit, keyword=keyword)
|
||||
usage_by_agent_id: dict[str, list[str]] = {}
|
||||
if app_id:
|
||||
draft_workflow = self._session.scalar(
|
||||
select(Workflow)
|
||||
.where(
|
||||
Workflow.tenant_id == tenant_id,
|
||||
Workflow.app_id == app_id,
|
||||
Workflow.version == Workflow.VERSION_DRAFT,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
if draft_workflow:
|
||||
agent_ids = [item["id"] for item in result["data"]]
|
||||
if agent_ids:
|
||||
bindings = self._session.scalars(
|
||||
select(WorkflowAgentNodeBinding).where(
|
||||
WorkflowAgentNodeBinding.tenant_id == tenant_id,
|
||||
WorkflowAgentNodeBinding.workflow_id == draft_workflow.id,
|
||||
WorkflowAgentNodeBinding.agent_id.in_(agent_ids),
|
||||
)
|
||||
).all()
|
||||
for binding in bindings:
|
||||
if binding.agent_id:
|
||||
usage_by_agent_id.setdefault(binding.agent_id, []).append(binding.node_id)
|
||||
|
||||
for item in result["data"]:
|
||||
existing_node_ids = usage_by_agent_id.get(item["id"], [])
|
||||
item["is_in_current_workflow"] = bool(existing_node_ids)
|
||||
item["in_current_workflow_count"] = len(existing_node_ids)
|
||||
item["existing_node_ids"] = existing_node_ids
|
||||
return result
|
||||
|
||||
def create_roster_agent(
|
||||
self,
|
||||
*,
|
||||
tenant_id: str,
|
||||
account_id: str,
|
||||
payload: RosterAgentCreatePayload,
|
||||
source: AgentSource = AgentSource.AGENT_APP,
|
||||
) -> Agent:
|
||||
ComposerConfigValidator.validate_agent_soul(payload.agent_soul)
|
||||
|
||||
agent = Agent(
|
||||
tenant_id=tenant_id,
|
||||
name=payload.name,
|
||||
description=payload.description,
|
||||
icon_type=payload.icon_type,
|
||||
icon=payload.icon,
|
||||
icon_background=payload.icon_background,
|
||||
agent_kind=AgentKind.DIFY_AGENT,
|
||||
scope=AgentScope.ROSTER,
|
||||
source=source,
|
||||
status=AgentStatus.ACTIVE,
|
||||
created_by=account_id,
|
||||
updated_by=account_id,
|
||||
)
|
||||
self._session.add(agent)
|
||||
try:
|
||||
self._session.flush()
|
||||
except IntegrityError as exc:
|
||||
self._session.rollback()
|
||||
raise AgentNameConflictError() from exc
|
||||
|
||||
version = AgentConfigSnapshot(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
version=1,
|
||||
config_snapshot=payload.agent_soul,
|
||||
version_note=payload.version_note,
|
||||
created_by=account_id,
|
||||
)
|
||||
self._session.add(version)
|
||||
self._session.flush()
|
||||
|
||||
revision = AgentConfigRevision(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
current_snapshot_id=version.id,
|
||||
revision=1,
|
||||
operation=AgentConfigRevisionOperation.CREATE_VERSION,
|
||||
version_note=payload.version_note,
|
||||
created_by=account_id,
|
||||
)
|
||||
self._session.add(revision)
|
||||
agent.active_config_snapshot_id = version.id
|
||||
|
||||
try:
|
||||
self._session.commit()
|
||||
except IntegrityError as exc:
|
||||
self._session.rollback()
|
||||
raise AgentNameConflictError() from exc
|
||||
return agent
|
||||
|
||||
def get_roster_agent_detail(self, *, tenant_id: str, agent_id: str) -> dict[str, Any]:
|
||||
agent = self._get_agent(tenant_id=tenant_id, agent_id=agent_id, roster_only=True)
|
||||
active_version = self._get_version(
|
||||
tenant_id=tenant_id, agent_id=agent.id, version_id=agent.active_config_snapshot_id
|
||||
)
|
||||
return self.serialize_agent(agent, active_version)
|
||||
|
||||
def update_roster_agent(
|
||||
self, *, tenant_id: str, agent_id: str, account_id: str, payload: RosterAgentUpdatePayload
|
||||
) -> dict[str, Any]:
|
||||
agent = self._get_agent(tenant_id=tenant_id, agent_id=agent_id, roster_only=True)
|
||||
if agent.status == AgentStatus.ARCHIVED:
|
||||
raise AgentArchivedError()
|
||||
|
||||
update_data = payload.model_dump(exclude_unset=True)
|
||||
for key, value in update_data.items():
|
||||
setattr(agent, key, value)
|
||||
agent.updated_by = account_id
|
||||
|
||||
try:
|
||||
self._session.commit()
|
||||
except IntegrityError as exc:
|
||||
self._session.rollback()
|
||||
raise AgentNameConflictError() from exc
|
||||
return self.get_roster_agent_detail(tenant_id=tenant_id, agent_id=agent_id)
|
||||
|
||||
def archive_roster_agent(self, *, tenant_id: str, agent_id: str, account_id: str) -> None:
|
||||
agent = self._get_agent(tenant_id=tenant_id, agent_id=agent_id, roster_only=True)
|
||||
if agent.status == AgentStatus.ARCHIVED:
|
||||
return
|
||||
agent.status = AgentStatus.ARCHIVED
|
||||
agent.archived_by = account_id
|
||||
agent.archived_at = naive_utc_now()
|
||||
agent.updated_by = account_id
|
||||
self._session.commit()
|
||||
|
||||
def list_agent_versions(self, *, tenant_id: str, agent_id: str) -> list[dict[str, Any]]:
|
||||
self._get_agent(tenant_id=tenant_id, agent_id=agent_id, roster_only=True)
|
||||
versions = list(
|
||||
self._session.scalars(
|
||||
select(AgentConfigSnapshot)
|
||||
.where(AgentConfigSnapshot.tenant_id == tenant_id, AgentConfigSnapshot.agent_id == agent_id)
|
||||
.order_by(AgentConfigSnapshot.version.desc())
|
||||
).all()
|
||||
)
|
||||
return [
|
||||
serialized_version
|
||||
for version in versions
|
||||
if (serialized_version := self.serialize_version(version)) is not None
|
||||
]
|
||||
|
||||
def get_agent_version_detail(self, *, tenant_id: str, agent_id: str, version_id: str) -> dict[str, Any]:
|
||||
self._get_agent(tenant_id=tenant_id, agent_id=agent_id, roster_only=True)
|
||||
version = self._get_version(tenant_id=tenant_id, agent_id=agent_id, version_id=version_id)
|
||||
revisions = list(
|
||||
self._session.scalars(
|
||||
select(AgentConfigRevision)
|
||||
.where(
|
||||
AgentConfigRevision.tenant_id == tenant_id,
|
||||
AgentConfigRevision.agent_id == agent_id,
|
||||
AgentConfigRevision.current_snapshot_id == version_id,
|
||||
)
|
||||
.order_by(AgentConfigRevision.revision.desc())
|
||||
).all()
|
||||
)
|
||||
result = self.serialize_version(version) or {}
|
||||
result["config_snapshot"] = version.config_snapshot_dict
|
||||
result["revisions"] = [
|
||||
{
|
||||
"id": revision.id,
|
||||
"previous_snapshot_id": revision.previous_snapshot_id,
|
||||
"current_snapshot_id": revision.current_snapshot_id,
|
||||
"revision": revision.revision,
|
||||
"operation": revision.operation.value,
|
||||
"summary": revision.summary,
|
||||
"version_note": revision.version_note,
|
||||
"created_by": revision.created_by,
|
||||
"created_at": revision.created_at.isoformat() if revision.created_at else None,
|
||||
}
|
||||
for revision in revisions
|
||||
]
|
||||
return result
|
||||
|
||||
def _get_agent(self, *, tenant_id: str, agent_id: str, roster_only: bool = False) -> Agent:
|
||||
stmt = select(Agent).where(Agent.tenant_id == tenant_id, Agent.id == agent_id)
|
||||
if roster_only:
|
||||
stmt = stmt.where(Agent.scope == AgentScope.ROSTER)
|
||||
agent = self._session.scalar(stmt.limit(1))
|
||||
if not agent:
|
||||
raise AgentNotFoundError()
|
||||
return agent
|
||||
|
||||
def _get_version(self, *, tenant_id: str, agent_id: str, version_id: str | None) -> AgentConfigSnapshot:
|
||||
if not version_id:
|
||||
raise AgentVersionNotFoundError()
|
||||
version = self._session.scalar(
|
||||
select(AgentConfigSnapshot)
|
||||
.where(
|
||||
AgentConfigSnapshot.tenant_id == tenant_id,
|
||||
AgentConfigSnapshot.agent_id == agent_id,
|
||||
AgentConfigSnapshot.id == version_id,
|
||||
)
|
||||
.limit(1)
|
||||
)
|
||||
if not version:
|
||||
raise AgentVersionNotFoundError()
|
||||
return version
|
||||
|
||||
def _load_versions_by_id(self, version_ids: list[str]) -> dict[str, AgentConfigSnapshot]:
|
||||
if not version_ids:
|
||||
return {}
|
||||
versions = self._session.scalars(
|
||||
select(AgentConfigSnapshot).where(AgentConfigSnapshot.id.in_(version_ids))
|
||||
).all()
|
||||
return {version.id: version for version in versions}
|
||||
@@ -0,0 +1,93 @@
|
||||
from enum import StrEnum
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
|
||||
from models.agent import AgentIconType
|
||||
from models.agent_config_entities import AgentSoulConfig, WorkflowNodeJobConfig
|
||||
|
||||
|
||||
class ComposerVariant(StrEnum):
|
||||
WORKFLOW = "workflow"
|
||||
AGENT_APP = "agent_app"
|
||||
|
||||
|
||||
class ComposerSaveStrategy(StrEnum):
|
||||
NODE_JOB_ONLY = "node_job_only"
|
||||
SAVE_TO_CURRENT_VERSION = "save_to_current_version"
|
||||
SAVE_AS_NEW_VERSION = "save_as_new_version"
|
||||
SAVE_AS_NEW_AGENT = "save_as_new_agent"
|
||||
SAVE_TO_ROSTER = "save_to_roster"
|
||||
|
||||
|
||||
class ComposerBindingPayload(BaseModel):
|
||||
binding_type: Literal["roster_agent", "inline_agent"]
|
||||
agent_id: str | None = None
|
||||
current_snapshot_id: str | None = None
|
||||
|
||||
|
||||
class ComposerSoulLockPayload(BaseModel):
|
||||
locked: bool = True
|
||||
unlocked_from_version_id: str | None = None
|
||||
|
||||
|
||||
class ComposerSavePayload(BaseModel):
|
||||
variant: ComposerVariant
|
||||
binding: ComposerBindingPayload | None = None
|
||||
soul_lock: ComposerSoulLockPayload = Field(default_factory=ComposerSoulLockPayload)
|
||||
agent_soul: AgentSoulConfig | None = None
|
||||
node_job: WorkflowNodeJobConfig | None = None
|
||||
save_strategy: ComposerSaveStrategy
|
||||
version_note: str | None = None
|
||||
idempotency_key: str | None = None
|
||||
client_revision_id: str | None = None
|
||||
new_agent_name: str | None = Field(default=None, min_length=1, max_length=255)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_variant_sections(self) -> "ComposerSavePayload":
|
||||
if self.variant == ComposerVariant.AGENT_APP and self.node_job is not None:
|
||||
raise ValueError("Agent App Variant must not include workflow node job config")
|
||||
if self.variant == ComposerVariant.AGENT_APP and self.agent_soul is not None:
|
||||
if self.agent_soul.app_variables and self.save_strategy == ComposerSaveStrategy.NODE_JOB_ONLY:
|
||||
raise ValueError("Agent App Variant cannot use node_job_only save strategy")
|
||||
if self.variant == ComposerVariant.WORKFLOW and self.agent_soul is not None:
|
||||
if self.agent_soul.app_variables:
|
||||
raise ValueError("Workflow Variant must not include app variables")
|
||||
if self.agent_soul.app_features:
|
||||
raise ValueError("Workflow Variant must not include app features")
|
||||
return self
|
||||
|
||||
|
||||
class RosterAgentCreatePayload(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=255)
|
||||
description: str = ""
|
||||
icon_type: AgentIconType | None = None
|
||||
icon: str | None = Field(default=None, max_length=255)
|
||||
icon_background: str | None = Field(default=None, max_length=255)
|
||||
agent_soul: AgentSoulConfig = Field(default_factory=AgentSoulConfig)
|
||||
version_note: str | None = None
|
||||
|
||||
|
||||
class RosterAgentUpdatePayload(BaseModel):
|
||||
name: str | None = Field(default=None, min_length=1, max_length=255)
|
||||
description: str | None = None
|
||||
icon_type: AgentIconType | None = None
|
||||
icon: str | None = Field(default=None, max_length=255)
|
||||
icon_background: str | None = Field(default=None, max_length=255)
|
||||
|
||||
|
||||
class RosterListQuery(BaseModel):
|
||||
page: int = Field(default=1, ge=1)
|
||||
limit: int = Field(default=20, ge=1, le=100)
|
||||
keyword: str | None = None
|
||||
|
||||
|
||||
class ComposerCandidateCapabilities(BaseModel):
|
||||
human_roster_available: bool = False
|
||||
|
||||
|
||||
class ComposerCandidatesResponse(BaseModel):
|
||||
variant: ComposerVariant
|
||||
allowed_node_job_candidates: dict[str, Any] = Field(default_factory=dict)
|
||||
allowed_soul_candidates: dict[str, Any] = Field(default_factory=dict)
|
||||
capabilities: ComposerCandidateCapabilities = Field(default_factory=ComposerCandidateCapabilities)
|
||||
Reference in New Issue
Block a user