from celery.result import AsyncResult from django.db import IntegrityError, transaction from django.utils import timezone from ninja import Router from ninja.errors import HttpError from apps.accounts.api import require_permission from apps.agents.models import ( AgentDefinition, AgentRun, AgentStepBinding, AgentVersion, ) from apps.agents.schemas import ( AgentCreateIn, AgentDefinitionOut, AgentDefinitionUpdateIn, AgentDetailOut, AgentDraftUpdateIn, AgentRunDetailOut, AgentRunIn, AgentRunSummaryOut, AgentVersionOut, ) from apps.agents.services import ( AgentValidationError, build_config_snapshot, ensure_agent_draft, replace_agent_steps, validate_agent_version, ) from apps.agents.tasks import execute_agent_run from apps.audit.models import OperationAuditLog from apps.skills.services import SkillValidationError, validate_payload router = Router(tags=["agents"]) runs_router = Router(tags=["agent-runs"]) def serialize_step(step: AgentStepBinding) -> dict: return { "id": step.id, "code": step.code, "order": step.order, "skill_version_id": step.skill_version_id, "skill_name": step.skill_version.definition.name, "skill_version": step.skill_version.version, "skill_status": step.skill_version.status, "input_mapping": step.input_mapping, } def serialize_version(version: AgentVersion) -> dict: return { "id": version.id, "version": version.version, "status": version.status, "model_deployment_id": version.model_deployment_id, "model_display_name": version.model_deployment.display_name, "system_prompt": version.system_prompt, "input_schema": version.input_schema, "example_input": version.example_input, "temperature_override": version.temperature_override, "max_output_tokens_override": version.max_output_tokens_override, "steps": [serialize_step(step) for step in version.steps.all()], "created_at": version.created_at, "updated_at": version.updated_at, "published_at": version.published_at, } def serialize_definition(definition: AgentDefinition, detailed: bool = False) -> dict: draft = definition.versions.filter(status="draft").first() payload = { "id": definition.id, "code": definition.code, "name": definition.name, "description": definition.description, "active_version": definition.active_version.version if definition.active_version_id else None, "draft_version": draft.version if draft else None, "updated_at": definition.updated_at, } if detailed: payload["versions"] = [serialize_version(version) for version in definition.versions.all()] return payload def serialize_run(run: AgentRun, detailed: bool = False) -> dict: payload = { "public_id": run.public_id, "agent_id": run.definition_id, "agent_name": run.definition.name, "agent_version": run.agent_version.version, "status": run.status, "sandbox": run.sandbox, "current_step": run.current_step, "total_steps": run.total_steps, "input_tokens": run.input_tokens, "output_tokens": run.output_tokens, "duration_ms": run.duration_ms, "error_code": run.error_code, "error_message": run.error_message, "created_at": run.created_at, "started_at": run.started_at, "finished_at": run.finished_at, } if detailed: payload.update( { "input_payload": run.input_payload, "final_output": run.final_output, "config_snapshot": run.config_snapshot, "cancellation_requested_at": run.cancellation_requested_at, "steps": [ { "id": step.id, "code": step.code, "name": step.name, "order": step.order, "status": step.status, "input_payload": step.input_payload, "output_payload": step.output_payload, "input_tokens": step.input_tokens, "output_tokens": step.output_tokens, "duration_ms": step.duration_ms, "error_code": step.error_code, "error_message": step.error_message, "attempts": list(step.attempts.all()), } for step in run.steps.all() ], } ) return payload def audit(user, action: str, target_type: str, target_id: str, snapshot: dict) -> None: OperationAuditLog.objects.create( user=user, module="agents", action=action, target_type=target_type, target_id=target_id, request_snapshot=snapshot, result="success", ) @router.get("", response=list[AgentDefinitionOut]) def list_agents(request): require_permission(request, "agent.manage") queryset = AgentDefinition.objects.select_related("active_version").prefetch_related("versions") return [serialize_definition(item) for item in queryset] @router.post("", response=AgentDetailOut) def create_agent(request, payload: AgentCreateIn): require_permission(request, "agent.manage") try: with transaction.atomic(): definition = AgentDefinition.objects.create( code=payload.code, name=payload.name, description=payload.description, created_by=request.user, ) version = AgentVersion.objects.create( definition=definition, version=1, model_deployment_id=payload.model_deployment_id, system_prompt=payload.system_prompt, input_schema=payload.input_schema, example_input=payload.example_input, temperature_override=payload.temperature_override, max_output_tokens_override=payload.max_output_tokens_override, created_by=request.user, ) replace_agent_steps(version, [item.dict() for item in payload.steps]) except (AgentValidationError, SkillValidationError) as error: raise HttpError(400, str(error)) from error except IntegrityError as error: raise HttpError(400, "Agent code or step configuration already exists") from error audit(request.user, "create", "agent", str(definition.id), {"code": definition.code}) definition = AgentDefinition.objects.select_related("active_version").get(id=definition.id) return serialize_definition(definition, detailed=True) @router.get("/{agent_id}", response=AgentDetailOut) def get_agent(request, agent_id: int): require_permission(request, "agent.manage") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) return serialize_definition(definition, detailed=True) @router.patch("/{agent_id}", response=AgentDetailOut) def update_agent(request, agent_id: int, payload: AgentDefinitionUpdateIn): require_permission(request, "agent.manage") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) values = payload.dict(exclude_unset=True) for field, value in values.items(): setattr(definition, field, value) definition.save() audit(request.user, "update", "agent", str(definition.id), {"fields": sorted(values)}) return serialize_definition(definition, detailed=True) @router.post("/{agent_id}/draft", response=AgentVersionOut) def create_agent_draft(request, agent_id: int): require_permission(request, "agent.manage") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) try: return serialize_version(ensure_agent_draft(definition, request.user)) except AgentValidationError as error: raise HttpError(400, str(error)) from error @router.get("/{agent_id}/draft", response=AgentVersionOut) def get_agent_draft(request, agent_id: int): require_permission(request, "agent.manage") version = AgentVersion.objects.select_related("model_deployment").get( definition_id=agent_id, status="draft", ) return serialize_version(version) @router.patch("/{agent_id}/draft", response=AgentVersionOut) def update_agent_draft(request, agent_id: int, payload: AgentDraftUpdateIn): require_permission(request, "agent.manage") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) try: version = ensure_agent_draft(definition, request.user) values = payload.dict(exclude_unset=True) steps = values.pop("steps", None) for field, value in values.items(): setattr(version, field, value) version.save() if steps is not None: replace_agent_steps(version, steps) else: validate_agent_version(version) except (AgentValidationError, SkillValidationError, IntegrityError) as error: raise HttpError(400, str(error)) from error audit( request.user, "update_draft", "agent", str(definition.id), {"version": version.version}, ) return serialize_version(version) @router.get("/{agent_id}/versions", response=list[AgentVersionOut]) def list_agent_versions(request, agent_id: int): require_permission(request, "agent.manage") versions = AgentVersion.objects.select_related("model_deployment").filter( definition_id=agent_id ) return [serialize_version(version) for version in versions] @router.post("/{agent_id}/publish", response=AgentVersionOut) def publish_agent(request, agent_id: int): require_permission(request, "agent.manage") with transaction.atomic(): definition = AgentDefinition.objects.select_for_update().get(id=agent_id) version = AgentVersion.objects.select_related("model_deployment").get( definition=definition, status="draft", ) try: validate_agent_version(version, publishing=True) except (AgentValidationError, SkillValidationError) as error: raise HttpError(400, str(error)) from error version.status = "published" version.published_by = request.user version.published_at = timezone.now() version.save(update_fields=["status", "published_by", "published_at", "updated_at"]) definition.active_version = version definition.save(update_fields=["active_version", "updated_at"]) audit(request.user, "publish", "agent", str(definition.id), {"version": version.version}) return serialize_version(version) @router.post("/{agent_id}/rollback/{version}", response=AgentDetailOut) def rollback_agent(request, agent_id: int, version: int): require_permission(request, "agent.manage") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) target = AgentVersion.objects.get( definition=definition, version=version, status="published", ) definition.active_version = target definition.save(update_fields=["active_version", "updated_at"]) audit(request.user, "rollback", "agent", str(definition.id), {"version": version}) return serialize_definition(definition, detailed=True) @router.post("/{agent_id}/runs", response=AgentRunDetailOut) def start_agent_run(request, agent_id: int, payload: AgentRunIn): require_permission(request, "agent.use") definition = AgentDefinition.objects.select_related("active_version").get(id=agent_id) if str(payload.version) == "draft": version = AgentVersion.objects.get(definition=definition, status="draft") sandbox = True else: version = AgentVersion.objects.get( definition=definition, version=int(payload.version), status="published", ) sandbox = False try: validate_agent_version(version, publishing=not sandbox) validate_payload(version.input_schema, payload.input, "Agent 输入") except (AgentValidationError, SkillValidationError) as error: raise HttpError(400, str(error)) from error run = AgentRun.objects.create( definition=definition, agent_version=version, sandbox=sandbox, input_payload=payload.input, config_snapshot=build_config_snapshot(version), total_steps=version.steps.count(), created_by=request.user, ) result = execute_agent_run.delay(run.id) run.refresh_from_db() if not run.task_id: run.task_id = result.id or "" run.save(update_fields=["task_id"]) audit( request.user, "run_started", "agent_run", str(run.public_id), {"agent_id": definition.id, "version": version.version, "sandbox": sandbox}, ) return serialize_run(run, detailed=True) @runs_router.get("", response=list[AgentRunSummaryOut]) def list_agent_runs(request, status: str | None = None, agent_id: int | None = None): require_permission(request, "agent.run.view") queryset = AgentRun.objects.select_related("definition", "agent_version") if status: queryset = queryset.filter(status=status) if agent_id: queryset = queryset.filter(definition_id=agent_id) return [serialize_run(run) for run in queryset[:100]] @runs_router.get("/{public_id}", response=AgentRunDetailOut) def get_agent_run(request, public_id: str): require_permission(request, "agent.run.view") run = AgentRun.objects.select_related("definition", "agent_version").get(public_id=public_id) return serialize_run(run, detailed=True) @runs_router.post("/{public_id}/cancel", response=AgentRunDetailOut) def cancel_agent_run(request, public_id: str): require_permission(request, "agent.use") run = AgentRun.objects.select_related("definition", "agent_version").get(public_id=public_id) if run.status not in {"queued", "running"}: raise HttpError(400, "Only queued or running runs can be cancelled") run.cancellation_requested_at = timezone.now() update_fields = ["cancellation_requested_at"] if run.status == "queued": run.status = "cancelled" run.finished_at = timezone.now() update_fields.extend(["status", "finished_at"]) run.save(update_fields=update_fields) if run.task_id: AsyncResult(run.task_id).revoke(terminate=False) audit(request.user, "cancel", "agent_run", str(run.public_id), {"status": run.status}) return serialize_run(run, detailed=True)