diff --git a/backend/apps/agents/__init__.py b/backend/apps/agents/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/apps/agents/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/apps/agents/api.py b/backend/apps/agents/api.py new file mode 100644 index 0000000..7d4dc78 --- /dev/null +++ b/backend/apps/agents/api.py @@ -0,0 +1,380 @@ +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) diff --git a/backend/apps/agents/apps.py b/backend/apps/agents/apps.py new file mode 100644 index 0000000..ed8d3e5 --- /dev/null +++ b/backend/apps/agents/apps.py @@ -0,0 +1,6 @@ +from django.apps import AppConfig + + +class AgentsConfig(AppConfig): + default_auto_field = "django.db.models.BigAutoField" + name = "apps.agents" diff --git a/backend/apps/agents/management/__init__.py b/backend/apps/agents/management/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/apps/agents/management/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/apps/agents/management/commands/__init__.py b/backend/apps/agents/management/commands/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/apps/agents/management/commands/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/apps/agents/management/commands/seed_ai_demo.py b/backend/apps/agents/management/commands/seed_ai_demo.py new file mode 100644 index 0000000..e5c5cd3 --- /dev/null +++ b/backend/apps/agents/management/commands/seed_ai_demo.py @@ -0,0 +1,194 @@ +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction +from django.utils import timezone + +from apps.agents.models import AgentDefinition, AgentStepBinding, AgentVersion +from apps.modelhub.models import ModelDeployment +from apps.skills.models import SkillDefinition, SkillVersion + +SKILLS = [ + { + "code": "issue-structure", + "name": "问题结构化", + "description": "将技术支持问题整理为统一结构。", + "prompt": ( + "请分析技术支持问题：{{ input.problem }}，环境：{{ input.environment }}。只输出 JSON。" + ), + "input_schema": { + "type": "object", + "properties": { + "problem": {"type": "string"}, + "environment": {"type": "string"}, + }, + "required": ["problem", "environment"], + }, + "output_schema": { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "symptoms": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["summary", "symptoms"], + }, + }, + { + "code": "troubleshooting", + "name": "排查建议", + "description": "基于结构化问题生成排查步骤。", + "prompt": ( + "根据问题摘要 {{ input.summary }} 和现象 {{ input.symptoms }} " + "生成排查建议，只输出 JSON。" + ), + "input_schema": { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "symptoms": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["summary", "symptoms"], + }, + "output_schema": { + "type": "object", + "properties": { + "likely_cause": {"type": "string"}, + "steps": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["likely_cause", "steps"], + }, + }, + { + "code": "customer-reply", + "name": "客户回复草稿", + "description": "将排查建议整理为客户可读回复。", + "prompt": ( + "原因：{{ input.likely_cause }}；步骤：{{ input.steps }}。生成客户回复，只输出 JSON。" + ), + "input_schema": { + "type": "object", + "properties": { + "likely_cause": {"type": "string"}, + "steps": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["likely_cause", "steps"], + }, + "output_schema": { + "type": "object", + "properties": {"reply": {"type": "string"}}, + "required": ["reply"], + }, + }, +] + + +class Command(BaseCommand): + help = "Seed the technical support analysis Agent and its published Skills." + + def add_arguments(self, parser): + parser.add_argument("--deployment-id", type=int) + + @transaction.atomic + def handle(self, *args, **options): + deployment_id = options.get("deployment_id") + deployments = ModelDeployment.objects.filter(enabled=True, provider__enabled=True) + deployment = ( + deployments.filter(id=deployment_id).first() if deployment_id else deployments.first() + ) + if deployment is None: + raise CommandError( + "Configure an enabled model deployment before seeding the demo Agent." + ) + + versions = {} + for spec in SKILLS: + definition, _ = SkillDefinition.objects.get_or_create( + code=spec["code"], + defaults={"name": spec["name"], "description": spec["description"]}, + ) + version = definition.versions.filter(version=1).first() + if version is None: + version = SkillVersion.objects.create( + definition=definition, + version=1, + status="published", + prompt_template=spec["prompt"], + input_schema=spec["input_schema"], + output_schema=spec["output_schema"], + example_input={}, + published_at=timezone.now(), + ) + definition.active_version = version + definition.save(update_fields=["active_version", "updated_at"]) + versions[spec["code"]] = version + + definition, _ = AgentDefinition.objects.get_or_create( + code="support-analysis", + defaults={ + "name": "技术支持分析 Agent", + "description": "结构化问题、生成排查建议和客户回复草稿。", + }, + ) + draft = definition.versions.filter(status="draft").first() + if draft is None: + next_version = ( + definition.versions.order_by("-version").values_list("version", flat=True).first() + or 0 + ) + 1 + draft = AgentVersion.objects.create( + definition=definition, + version=next_version, + status="draft", + model_deployment=deployment, + system_prompt="你是企业技术支持助手。输出必须严格符合当前步骤要求的 JSON。", + input_schema={ + "type": "object", + "properties": { + "problem": {"type": "string"}, + "environment": {"type": "string"}, + }, + "required": ["problem", "environment"], + }, + example_input={ + "problem": "采集任务长时间没有更新", + "environment": "生产环境", + }, + ) + AgentStepBinding.objects.bulk_create( + [ + AgentStepBinding( + agent_version=draft, + code="structure", + order=1, + skill_version=versions["issue-structure"], + input_mapping={ + "problem": "$input.problem", + "environment": "$input.environment", + }, + ), + AgentStepBinding( + agent_version=draft, + code="troubleshooting", + order=2, + skill_version=versions["troubleshooting"], + input_mapping={ + "summary": "$steps.structure.output.summary", + "symptoms": "$steps.structure.output.symptoms", + }, + ), + AgentStepBinding( + agent_version=draft, + code="reply", + order=3, + skill_version=versions["customer-reply"], + input_mapping={ + "likely_cause": "$steps.troubleshooting.output.likely_cause", + "steps": "$steps.troubleshooting.output.steps", + }, + ), + ] + ) + self.stdout.write( + self.style.SUCCESS( + f"Seeded Agent {definition.code} draft v{draft.version} " + f"with deployment {deployment.id}." + ) + ) diff --git a/backend/apps/agents/migrations/0001_initial.py b/backend/apps/agents/migrations/0001_initial.py new file mode 100644 index 0000000..a914f60 --- /dev/null +++ b/backend/apps/agents/migrations/0001_initial.py @@ -0,0 +1,339 @@ +# Generated by Django 5.2.15 on 2026-07-14 05:54 + +import uuid + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + initial = True + + dependencies = [ + ("modelhub", "0001_initial"), + ("skills", "0001_initial"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="AgentDefinition", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("code", models.SlugField(max_length=80, unique=True)), + ("name", models.CharField(max_length=120)), + ("description", models.TextField(blank=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ( + "created_by", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="created_agents", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "ordering": ["name"], + }, + ), + migrations.CreateModel( + name="AgentRun", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("public_id", models.UUIDField(default=uuid.uuid4, editable=False, unique=True)), + ("sandbox", models.BooleanField(default=False)), + ("status", models.CharField(default="queued", max_length=24)), + ("input_payload", models.JSONField(default=dict)), + ("final_output", models.JSONField(blank=True, null=True)), + ("config_snapshot", models.JSONField(default=dict)), + ("current_step", models.CharField(blank=True, max_length=80)), + ("total_steps", models.PositiveIntegerField(default=0)), + ("input_tokens", models.PositiveIntegerField(default=0)), + ("output_tokens", models.PositiveIntegerField(default=0)), + ("duration_ms", models.PositiveIntegerField(default=0)), + ("task_id", models.CharField(blank=True, max_length=120)), + ("cancellation_requested_at", models.DateTimeField(blank=True, null=True)), + ("error_code", models.CharField(blank=True, max_length=80)), + ("error_message", models.TextField(blank=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("started_at", models.DateTimeField(blank=True, null=True)), + ("finished_at", models.DateTimeField(blank=True, null=True)), + ( + "created_by", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="agent_runs", + to=settings.AUTH_USER_MODEL, + ), + ), + ( + "definition", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="runs", + to="agents.agentdefinition", + ), + ), + ], + options={ + "ordering": ["-created_at"], + }, + ), + migrations.CreateModel( + name="AgentRunStep", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("code", models.CharField(max_length=80)), + ("name", models.CharField(max_length=120)), + ("order", models.PositiveIntegerField()), + ("status", models.CharField(default="queued", max_length=24)), + ("input_payload", models.JSONField(default=dict)), + ("output_payload", models.JSONField(blank=True, null=True)), + ("input_tokens", models.PositiveIntegerField(default=0)), + ("output_tokens", models.PositiveIntegerField(default=0)), + ("duration_ms", models.PositiveIntegerField(default=0)), + ("error_code", models.CharField(blank=True, max_length=80)), + ("error_message", models.TextField(blank=True)), + ("started_at", models.DateTimeField(blank=True, null=True)), + ("finished_at", models.DateTimeField(blank=True, null=True)), + ( + "run", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="steps", + to="agents.agentrun", + ), + ), + ( + "skill_version", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="run_steps", + to="skills.skillversion", + ), + ), + ], + options={ + "ordering": ["order"], + }, + ), + migrations.CreateModel( + name="AgentRunAttempt", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("attempt_type", models.CharField(default="initial", max_length=32)), + ("status", models.CharField(max_length=24)), + ("raw_output", models.TextField(blank=True)), + ("validation_error", models.TextField(blank=True)), + ("input_tokens", models.PositiveIntegerField(default=0)), + ("output_tokens", models.PositiveIntegerField(default=0)), + ("duration_ms", models.PositiveIntegerField(default=0)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ( + "model_invocation", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="agent_attempts", + to="modelhub.modelinvocation", + ), + ), + ( + "step", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="attempts", + to="agents.agentrunstep", + ), + ), + ], + options={ + "ordering": ["created_at"], + }, + ), + migrations.CreateModel( + name="AgentVersion", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("version", models.PositiveIntegerField()), + ("status", models.CharField(default="draft", max_length=24)), + ("system_prompt", models.TextField(blank=True)), + ("input_schema", models.JSONField(default=dict)), + ("example_input", models.JSONField(blank=True, default=dict)), + ("temperature_override", models.FloatField(blank=True, null=True)), + ("max_output_tokens_override", models.PositiveIntegerField(blank=True, null=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ("published_at", models.DateTimeField(blank=True, null=True)), + ( + "created_by", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="created_agent_versions", + to=settings.AUTH_USER_MODEL, + ), + ), + ( + "definition", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="versions", + to="agents.agentdefinition", + ), + ), + ( + "model_deployment", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="agent_versions", + to="modelhub.modeldeployment", + ), + ), + ( + "published_by", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="published_agent_versions", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "ordering": ["-version"], + }, + ), + migrations.CreateModel( + name="AgentStepBinding", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, primary_key=True, serialize=False, verbose_name="ID" + ), + ), + ("code", models.SlugField(max_length=80)), + ("order", models.PositiveIntegerField()), + ("input_mapping", models.JSONField(default=dict)), + ( + "skill_version", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="agent_bindings", + to="skills.skillversion", + ), + ), + ( + "agent_version", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="steps", + to="agents.agentversion", + ), + ), + ], + options={ + "ordering": ["order"], + }, + ), + migrations.AddField( + model_name="agentrun", + name="agent_version", + field=models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="runs", + to="agents.agentversion", + ), + ), + migrations.AddField( + model_name="agentdefinition", + name="active_version", + field=models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.SET_NULL, + related_name="active_for_definitions", + to="agents.agentversion", + ), + ), + migrations.AddConstraint( + model_name="agentrunstep", + constraint=models.UniqueConstraint( + fields=("run", "order"), name="unique_run_step_order" + ), + ), + migrations.AddConstraint( + model_name="agentversion", + constraint=models.UniqueConstraint( + fields=("definition", "version"), name="unique_agent_version" + ), + ), + migrations.AddConstraint( + model_name="agentversion", + constraint=models.UniqueConstraint( + condition=models.Q(("status", "draft")), + fields=("definition",), + name="unique_agent_draft", + ), + ), + migrations.AddConstraint( + model_name="agentstepbinding", + constraint=models.UniqueConstraint( + fields=("agent_version", "code"), name="unique_agent_step_code" + ), + ), + migrations.AddConstraint( + model_name="agentstepbinding", + constraint=models.UniqueConstraint( + fields=("agent_version", "order"), name="unique_agent_step_order" + ), + ), + migrations.AddIndex( + model_name="agentrun", + index=models.Index( + fields=["status", "created_at"], name="agents_agen_status_27cafa_idx" + ), + ), + migrations.AddIndex( + model_name="agentrun", + index=models.Index( + fields=["definition", "created_at"], name="agents_agen_definit_22ef17_idx" + ), + ), + ] diff --git a/backend/apps/agents/migrations/__init__.py b/backend/apps/agents/migrations/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/apps/agents/migrations/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/apps/agents/models.py b/backend/apps/agents/models.py new file mode 100644 index 0000000..a356d64 --- /dev/null +++ b/backend/apps/agents/models.py @@ -0,0 +1,212 @@ +import uuid + +from django.conf import settings +from django.db import models +from django.db.models import Q + +from apps.modelhub.models import ModelDeployment, ModelInvocation +from apps.skills.models import SkillVersion + + +class AgentDefinition(models.Model): + code = models.SlugField(max_length=80, unique=True) + name = models.CharField(max_length=120) + description = models.TextField(blank=True) + active_version = models.ForeignKey( + "AgentVersion", + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="active_for_definitions", + ) + created_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="created_agents", + ) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + ordering = ["name"] + + +class AgentVersion(models.Model): + definition = models.ForeignKey( + AgentDefinition, + on_delete=models.CASCADE, + related_name="versions", + ) + version = models.PositiveIntegerField() + status = models.CharField(max_length=24, default="draft") + model_deployment = models.ForeignKey( + ModelDeployment, + on_delete=models.PROTECT, + related_name="agent_versions", + ) + system_prompt = models.TextField(blank=True) + input_schema = models.JSONField(default=dict) + example_input = models.JSONField(default=dict, blank=True) + temperature_override = models.FloatField(null=True, blank=True) + max_output_tokens_override = models.PositiveIntegerField(null=True, blank=True) + created_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="created_agent_versions", + ) + published_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="published_agent_versions", + ) + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + published_at = models.DateTimeField(null=True, blank=True) + + class Meta: + ordering = ["-version"] + constraints = [ + models.UniqueConstraint( + fields=["definition", "version"], + name="unique_agent_version", + ), + models.UniqueConstraint( + fields=["definition"], + condition=Q(status="draft"), + name="unique_agent_draft", + ), + ] + + +class AgentStepBinding(models.Model): + agent_version = models.ForeignKey( + AgentVersion, + on_delete=models.CASCADE, + related_name="steps", + ) + code = models.SlugField(max_length=80) + order = models.PositiveIntegerField() + skill_version = models.ForeignKey( + SkillVersion, + on_delete=models.PROTECT, + related_name="agent_bindings", + ) + input_mapping = models.JSONField(default=dict) + + class Meta: + ordering = ["order"] + constraints = [ + models.UniqueConstraint( + fields=["agent_version", "code"], + name="unique_agent_step_code", + ), + models.UniqueConstraint( + fields=["agent_version", "order"], + name="unique_agent_step_order", + ), + ] + + +class AgentRun(models.Model): + public_id = models.UUIDField(default=uuid.uuid4, unique=True, editable=False) + definition = models.ForeignKey( + AgentDefinition, + on_delete=models.PROTECT, + related_name="runs", + ) + agent_version = models.ForeignKey( + AgentVersion, + on_delete=models.PROTECT, + related_name="runs", + ) + sandbox = models.BooleanField(default=False) + status = models.CharField(max_length=24, default="queued") + input_payload = models.JSONField(default=dict) + final_output = models.JSONField(null=True, blank=True) + config_snapshot = models.JSONField(default=dict) + current_step = models.CharField(max_length=80, blank=True) + total_steps = models.PositiveIntegerField(default=0) + input_tokens = models.PositiveIntegerField(default=0) + output_tokens = models.PositiveIntegerField(default=0) + duration_ms = models.PositiveIntegerField(default=0) + task_id = models.CharField(max_length=120, blank=True) + cancellation_requested_at = models.DateTimeField(null=True, blank=True) + error_code = models.CharField(max_length=80, blank=True) + error_message = models.TextField(blank=True) + created_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="agent_runs", + ) + created_at = models.DateTimeField(auto_now_add=True) + started_at = models.DateTimeField(null=True, blank=True) + finished_at = models.DateTimeField(null=True, blank=True) + + class Meta: + ordering = ["-created_at"] + indexes = [ + models.Index(fields=["status", "created_at"]), + models.Index(fields=["definition", "created_at"]), + ] + + +class AgentRunStep(models.Model): + run = models.ForeignKey(AgentRun, on_delete=models.CASCADE, related_name="steps") + code = models.CharField(max_length=80) + name = models.CharField(max_length=120) + order = models.PositiveIntegerField() + skill_version = models.ForeignKey( + SkillVersion, + on_delete=models.PROTECT, + related_name="run_steps", + ) + status = models.CharField(max_length=24, default="queued") + input_payload = models.JSONField(default=dict) + output_payload = models.JSONField(null=True, blank=True) + input_tokens = models.PositiveIntegerField(default=0) + output_tokens = models.PositiveIntegerField(default=0) + duration_ms = models.PositiveIntegerField(default=0) + error_code = models.CharField(max_length=80, blank=True) + error_message = models.TextField(blank=True) + started_at = models.DateTimeField(null=True, blank=True) + finished_at = models.DateTimeField(null=True, blank=True) + + class Meta: + ordering = ["order"] + constraints = [ + models.UniqueConstraint(fields=["run", "order"], name="unique_run_step_order") + ] + + +class AgentRunAttempt(models.Model): + step = models.ForeignKey( + AgentRunStep, + on_delete=models.CASCADE, + related_name="attempts", + ) + model_invocation = models.ForeignKey( + ModelInvocation, + null=True, + blank=True, + on_delete=models.SET_NULL, + related_name="agent_attempts", + ) + attempt_type = models.CharField(max_length=32, default="initial") + status = models.CharField(max_length=24) + raw_output = models.TextField(blank=True) + validation_error = models.TextField(blank=True) + input_tokens = models.PositiveIntegerField(default=0) + output_tokens = models.PositiveIntegerField(default=0) + duration_ms = models.PositiveIntegerField(default=0) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + ordering = ["created_at"] diff --git a/backend/apps/agents/schemas.py b/backend/apps/agents/schemas.py new file mode 100644 index 0000000..f3ee65f --- /dev/null +++ b/backend/apps/agents/schemas.py @@ -0,0 +1,141 @@ +from datetime import datetime +from typing import Any +from uuid import UUID + +from ninja import Schema + + +class AgentStepIn(Schema): + code: str + skill_version_id: int + input_mapping: dict[str, str] + + +class AgentCreateIn(Schema): + code: str + name: str + description: str = "" + model_deployment_id: int + system_prompt: str = "" + input_schema: dict[str, Any] = {"type": "object"} + example_input: dict[str, Any] = {} + temperature_override: float | None = None + max_output_tokens_override: int | None = None + steps: list[AgentStepIn] + + +class AgentDefinitionUpdateIn(Schema): + name: str | None = None + description: str | None = None + + +class AgentDraftUpdateIn(Schema): + model_deployment_id: int | None = None + system_prompt: str | None = None + input_schema: dict[str, Any] | None = None + example_input: dict[str, Any] | None = None + temperature_override: float | None = None + max_output_tokens_override: int | None = None + steps: list[AgentStepIn] | None = None + + +class AgentStepOut(Schema): + id: int + code: str + order: int + skill_version_id: int + skill_name: str + skill_version: int + skill_status: str + input_mapping: dict[str, str] + + +class AgentVersionOut(Schema): + id: int + version: int + status: str + model_deployment_id: int + model_display_name: str + system_prompt: str + input_schema: dict[str, Any] + example_input: dict[str, Any] + temperature_override: float | None + max_output_tokens_override: int | None + steps: list[AgentStepOut] + created_at: datetime + updated_at: datetime + published_at: datetime | None + + +class AgentDefinitionOut(Schema): + id: int + code: str + name: str + description: str + active_version: int | None + draft_version: int | None + updated_at: datetime + + +class AgentDetailOut(AgentDefinitionOut): + versions: list[AgentVersionOut] + + +class AgentRunIn(Schema): + version: str | int = "draft" + input: dict[str, Any] + + +class AgentRunAttemptOut(Schema): + id: int + attempt_type: str + status: str + raw_output: str + validation_error: str + input_tokens: int + output_tokens: int + duration_ms: int + created_at: datetime + + +class AgentRunStepOut(Schema): + id: int + code: str + name: str + order: int + status: str + input_payload: dict[str, Any] + output_payload: Any | None + input_tokens: int + output_tokens: int + duration_ms: int + error_code: str + error_message: str + attempts: list[AgentRunAttemptOut] + + +class AgentRunSummaryOut(Schema): + public_id: UUID + agent_id: int + agent_name: str + agent_version: int + status: str + sandbox: bool + current_step: str + total_steps: int + input_tokens: int + output_tokens: int + duration_ms: int + error_code: str + error_message: str + created_at: datetime + started_at: datetime | None + finished_at: datetime | None + + +class AgentRunDetailOut(AgentRunSummaryOut): + input_payload: dict[str, Any] + final_output: Any | None + config_snapshot: dict[str, Any] + cancellation_requested_at: datetime | None + steps: list[AgentRunStepOut] diff --git a/backend/apps/agents/services.py b/backend/apps/agents/services.py new file mode 100644 index 0000000..4f5ad0b --- /dev/null +++ b/backend/apps/agents/services.py @@ -0,0 +1,390 @@ +import json +import re +from copy import deepcopy +from time import monotonic +from typing import Any + +from django.db import transaction +from django.utils import timezone + +from apps.agents.models import ( + AgentDefinition, + AgentRun, + AgentRunAttempt, + AgentRunStep, + AgentStepBinding, + AgentVersion, +) +from apps.audit.models import OperationAuditLog +from apps.modelhub.models import ModelInvocation +from apps.modelhub.services import execute_invocation +from apps.skills.services import ( + SkillValidationError, + render_prompt, + validate_json_schema, + validate_payload, +) + +SOURCE_PATH = re.compile( + r"^\$(?:input(?:\.[A-Za-z_][A-Za-z0-9_]*)*|steps\.[A-Za-z_][A-Za-z0-9_-]*\.output(?:\.[A-Za-z_][A-Za-z0-9_]*)*)$" +) + + +class AgentValidationError(ValueError): + pass + + +def validate_agent_version(version: AgentVersion, *, publishing: bool = False) -> None: + validate_json_schema(version.input_schema, "Agent 输入 Schema") + if version.example_input: + validate_payload(version.input_schema, version.example_input, "Agent 示例输入") + if version.temperature_override is not None and not 0 <= version.temperature_override <= 2: + raise AgentValidationError("Temperature 必须在 0 到 2 之间") + if ( + version.max_output_tokens_override is not None + and not 1 <= version.max_output_tokens_override <= 32768 + ): + raise AgentValidationError("最大输出 Token 超出允许范围") + steps = list(version.steps.select_related("skill_version__definition")) + if not steps: + raise AgentValidationError("Agent 至少需要一个 Skill 步骤") + if len(steps) > 10: + raise AgentValidationError("Agent 最多允许 10 个步骤") + known_steps: set[str] = set() + for step in steps: + if publishing and step.skill_version.status != "published": + raise AgentValidationError(f"步骤 {step.code} 必须绑定已发布 Skill") + for target, source in step.input_mapping.items(): + if not isinstance(target, str) or not target: + raise AgentValidationError(f"步骤 {step.code} 的映射字段无效") + if not isinstance(source, str) or not SOURCE_PATH.fullmatch(source): + raise AgentValidationError(f"步骤 {step.code} 的映射来源无效: {source}") + if source.startswith("$steps."): + source_step = source.split(".")[1] + if source_step not in known_steps: + raise AgentValidationError( + f"步骤 {step.code} 只能引用之前步骤的输出: {source_step}" + ) + known_steps.add(step.code) + + +def _resolve_source(path: str, run_input: dict, step_outputs: dict[str, dict]) -> Any: + parts = path.removeprefix("$").split(".") + if parts[0] == "input": + value: Any = run_input + tail = parts[1:] + else: + value = step_outputs.get(parts[1], {}).get("output") + tail = parts[3:] + for part in tail: + if not isinstance(value, dict) or part not in value: + raise AgentValidationError(f"映射来源不存在: {path}") + value = value[part] + return deepcopy(value) + + +def map_step_input( + binding: AgentStepBinding, + run_input: dict, + step_outputs: dict[str, dict], +) -> dict: + return { + target: _resolve_source(source, run_input, step_outputs) + for target, source in binding.input_mapping.items() + } + + +def ensure_agent_draft(definition: AgentDefinition, user=None) -> AgentVersion: + existing = definition.versions.filter(status="draft").first() + if existing: + return existing + source = definition.active_version + if source is None: + raise AgentValidationError("Agent 没有可克隆的已发布版本") + latest = definition.versions.order_by("-version").values_list("version", flat=True).first() or 0 + draft = AgentVersion.objects.create( + definition=definition, + version=latest + 1, + status="draft", + model_deployment=source.model_deployment, + system_prompt=source.system_prompt, + input_schema=deepcopy(source.input_schema), + example_input=deepcopy(source.example_input), + temperature_override=source.temperature_override, + max_output_tokens_override=source.max_output_tokens_override, + created_by=user, + ) + AgentStepBinding.objects.bulk_create( + [ + AgentStepBinding( + agent_version=draft, + code=step.code, + order=step.order, + skill_version=step.skill_version, + input_mapping=deepcopy(step.input_mapping), + ) + for step in source.steps.all() + ] + ) + return draft + + +def build_config_snapshot(version: AgentVersion) -> dict: + deployment = version.model_deployment + return { + "agent": { + "id": version.definition_id, + "code": version.definition.code, + "version": version.version, + "status": version.status, + }, + "model": { + "deployment_id": deployment.id, + "provider_name": deployment.provider.name, + "model_name": deployment.model_name, + "temperature": version.temperature_override + if version.temperature_override is not None + else deployment.default_temperature, + "max_output_tokens": version.max_output_tokens_override or deployment.max_output_tokens, + }, + "steps": [ + { + "code": step.code, + "order": step.order, + "skill_id": step.skill_version.definition_id, + "skill_code": step.skill_version.definition.code, + "skill_version": step.skill_version.version, + "input_mapping": step.input_mapping, + } + for step in version.steps.select_related("skill_version__definition") + ], + } + + +def _parse_json_output(raw_output: str) -> Any: + value = raw_output.strip() + if value.startswith("```"): + value = re.sub(r"^```(?:json)?\s*", "", value) + value = re.sub(r"\s*```$", "", value) + try: + return json.loads(value) + except json.JSONDecodeError as error: + raise SkillValidationError(f"模型输出不是合法 JSON: {error.msg}") from error + + +def _record_attempt( + step: AgentRunStep, + invocation: ModelInvocation, + attempt_type: str, + validation_error: str = "", +) -> AgentRunAttempt: + return AgentRunAttempt.objects.create( + step=step, + model_invocation=invocation, + attempt_type=attempt_type, + status=invocation.status, + raw_output=invocation.output_text, + validation_error=validation_error, + input_tokens=invocation.input_tokens or 0, + output_tokens=invocation.output_tokens or 0, + duration_ms=invocation.duration_ms, + ) + + +def _invoke_for_step( + run: AgentRun, + step: AgentRunStep, + prompt: str, + attempt_type: str, +) -> ModelInvocation: + version = run.agent_version + messages = [] + if version.system_prompt: + messages.append({"role": "system", "content": version.system_prompt}) + messages.append({"role": "user", "content": prompt}) + invocation = ModelInvocation.objects.create( + deployment=version.model_deployment, + invocation_type="agent_step", + input_snapshot={"messages": messages, "run_id": str(run.public_id), "step": step.code}, + created_by=run.created_by, + ) + execute_invocation( + invocation, + messages, + temperature=version.temperature_override, + max_output_tokens=version.max_output_tokens_override, + ) + _record_attempt(step, invocation, attempt_type) + if invocation.retry_count: + AgentRunAttempt.objects.create( + step=step, + model_invocation=invocation, + attempt_type="transient_retry", + status=invocation.status, + raw_output=invocation.output_text, + input_tokens=0, + output_tokens=0, + duration_ms=0, + ) + return invocation + + +def _repair_output( + run: AgentRun, + step: AgentRunStep, + raw_output: str, + validation_error: str, + output_schema: dict, +) -> ModelInvocation: + prompt = ( + "请只输出修正后的 JSON，不要解释。\n" + f"输出 Schema：{json.dumps(output_schema, ensure_ascii=False)}\n" + f"校验错误：{validation_error}\n" + f"原始输出：{raw_output}" + ) + invocation = _invoke_for_step(run, step, prompt, "repair") + attempt = step.attempts.order_by("-created_at").first() + if attempt: + attempt.validation_error = validation_error + attempt.save(update_fields=["validation_error"]) + return invocation + + +def execute_agent_run(run_id: int) -> AgentRun: + run = AgentRun.objects.select_related( + "definition", + "agent_version__model_deployment__provider", + "created_by", + ).get(id=run_id) + started = monotonic() + run.status = "running" + run.started_at = timezone.now() + run.save(update_fields=["status", "started_at"]) + step_outputs: dict[str, dict] = {} + try: + bindings = list( + run.agent_version.steps.select_related( + "skill_version__definition", + ) + ) + for binding in bindings: + run.refresh_from_db(fields=["cancellation_requested_at"]) + if run.cancellation_requested_at: + run.status = "cancelled" + break + run.current_step = binding.code + run.save(update_fields=["current_step"]) + step = AgentRunStep.objects.create( + run=run, + code=binding.code, + name=binding.skill_version.definition.name, + order=binding.order, + skill_version=binding.skill_version, + status="running", + started_at=timezone.now(), + ) + step_started = monotonic() + try: + step_input = map_step_input(binding, run.input_payload, step_outputs) + validate_payload( + binding.skill_version.input_schema, + step_input, + f"步骤 {binding.code} 输入", + ) + step.input_payload = step_input + step.save(update_fields=["input_payload"]) + prompt = render_prompt(binding.skill_version.prompt_template, step_input) + invocation = _invoke_for_step(run, step, prompt, "initial") + if invocation.status != "succeeded": + raise AgentValidationError(invocation.error_message or "模型调用失败") + try: + output = _parse_json_output(invocation.output_text) + validate_payload(binding.skill_version.output_schema, output, "模型输出") + except SkillValidationError as error: + repair = _repair_output( + run, + step, + invocation.output_text, + str(error), + binding.skill_version.output_schema, + ) + if repair.status != "succeeded": + raise AgentValidationError( + repair.error_message or "结构修复失败" + ) from error + output = _parse_json_output(repair.output_text) + validate_payload(binding.skill_version.output_schema, output, "修复输出") + step.output_payload = output + step.status = "succeeded" + step_outputs[binding.code] = {"output": output} + except (AgentValidationError, SkillValidationError) as error: + step.status = "failed" + step.error_code = "step_failed" + step.error_message = str(error) + run.status = "failed" + run.error_code = "step_failed" + run.error_message = f"步骤 {binding.code} 失败: {error}" + finally: + attempts = step.attempts.all() + step.input_tokens = sum(item.input_tokens for item in attempts) + step.output_tokens = sum(item.output_tokens for item in attempts) + step.duration_ms = max(int((monotonic() - step_started) * 1000), 0) + step.finished_at = timezone.now() + step.save() + if run.status == "failed": + break + run.refresh_from_db(fields=["cancellation_requested_at"]) + if run.cancellation_requested_at: + run.status = "cancelled" + break + if run.status == "running": + run.status = "succeeded" + run.final_output = step.output_payload if bindings else None + except Exception as error: + run.status = "failed" + run.error_code = "runtime_error" + run.error_message = "Agent 运行发生未预期错误" + OperationAuditLog.objects.create( + user=run.created_by, + module="agents", + action="runtime_error", + target_type="agent_run", + target_id=str(run.public_id), + request_snapshot={"error_type": type(error).__name__}, + result="failed", + ) + run.input_tokens = sum(item.input_tokens for item in run.steps.all()) + run.output_tokens = sum(item.output_tokens for item in run.steps.all()) + run.duration_ms = max(int((monotonic() - started) * 1000), 0) + run.finished_at = timezone.now() + run.save() + OperationAuditLog.objects.create( + user=run.created_by, + module="agents", + action="run_finished", + target_type="agent_run", + target_id=str(run.public_id), + request_snapshot={"status": run.status, "duration_ms": run.duration_ms}, + result="success" if run.status == "succeeded" else run.status, + ) + return run + + +def replace_agent_steps(version: AgentVersion, steps: list[dict]) -> None: + if len(steps) > 10: + raise AgentValidationError("Agent 最多允许 10 个步骤") + with transaction.atomic(): + version.steps.all().delete() + AgentStepBinding.objects.bulk_create( + [ + AgentStepBinding( + agent_version=version, + code=item["code"], + order=index + 1, + skill_version_id=item["skill_version_id"], + input_mapping=item.get("input_mapping", {}), + ) + for index, item in enumerate(steps) + ] + ) + validate_agent_version(version) diff --git a/backend/apps/agents/tasks.py b/backend/apps/agents/tasks.py new file mode 100644 index 0000000..e3b44f6 --- /dev/null +++ b/backend/apps/agents/tasks.py @@ -0,0 +1,55 @@ +from datetime import timedelta + +from celery import shared_task +from celery.exceptions import SoftTimeLimitExceeded +from django.conf import settings +from django.utils import timezone + +from apps.agents.models import AgentRun +from apps.agents.services import execute_agent_run as execute_agent_run_service +from apps.audit.models import OperationAuditLog + + +@shared_task( + bind=True, + name="apps.agents.tasks.execute_agent_run", + soft_time_limit=600, + time_limit=630, +) +def execute_agent_run(self, run_id: int) -> None: + try: + execute_agent_run_service(run_id) + except SoftTimeLimitExceeded: + run = AgentRun.objects.get(id=run_id) + run.status = "timed_out" + run.error_code = "run_timeout" + run.error_message = "Agent 运行超过 10 分钟限制" + run.finished_at = timezone.now() + run.save( + update_fields=[ + "status", + "error_code", + "error_message", + "finished_at", + ] + ) + + +@shared_task(name="apps.agents.tasks.cleanup_expired_agent_runs") +def cleanup_expired_agent_runs() -> int: + cutoff = timezone.now() - timedelta(days=settings.AI_RUN_RETENTION_DAYS) + queryset = AgentRun.objects.filter(created_at__lt=cutoff) + count = queryset.count() + if count: + OperationAuditLog.objects.create( + module="agents", + action="cleanup_runs", + target_type="agent_run", + request_snapshot={ + "deleted_count": count, + "retention_days": settings.AI_RUN_RETENTION_DAYS, + }, + result="success", + ) + queryset.delete() + return count diff --git a/backend/apps/agents/tests/__init__.py b/backend/apps/agents/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/apps/agents/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/apps/agents/tests/helpers.py b/backend/apps/agents/tests/helpers.py new file mode 100644 index 0000000..16c6549 --- /dev/null +++ b/backend/apps/agents/tests/helpers.py @@ -0,0 +1,91 @@ +from apps.agents.models import AgentDefinition, AgentStepBinding, AgentVersion +from apps.modelhub.crypto import encrypt_secret +from apps.modelhub.models import ModelDeployment, ModelProvider +from apps.skills.models import SkillDefinition, SkillVersion + + +def create_deployment(): + provider = ModelProvider.objects.create( + name="模型渠道", + base_url="https://models.example.com/v1", + api_key_encrypted=encrypt_secret("secret"), + ) + return ModelDeployment.objects.create( + provider=provider, + display_name="支持模型", + model_name="support-model", + ) + + +def create_skill(code: str, output_schema: dict, input_schema: dict | None = None): + definition = SkillDefinition.objects.create(code=code, name=code) + version = SkillVersion.objects.create( + definition=definition, + version=1, + status="published", + prompt_template="处理：{{ input.value }}", + input_schema=input_schema + or { + "type": "object", + "properties": {"value": {}}, + "required": ["value"], + }, + output_schema=output_schema, + ) + definition.active_version = version + definition.save(update_fields=["active_version"]) + return version + + +def create_agent(deployment=None, *, published=False, two_steps=True): + deployment = deployment or create_deployment() + first_skill = create_skill( + "first-skill", + { + "type": "object", + "properties": {"summary": {"type": "string"}}, + "required": ["summary"], + }, + ) + second_skill = None + if two_steps: + second_skill = create_skill( + "second-skill", + { + "type": "object", + "properties": {"reply": {"type": "string"}}, + "required": ["reply"], + }, + ) + definition = AgentDefinition.objects.create(code="support-agent", name="支持 Agent") + version = AgentVersion.objects.create( + definition=definition, + version=1, + status="published" if published else "draft", + model_deployment=deployment, + input_schema={ + "type": "object", + "properties": {"problem": {"type": "string"}}, + "required": ["problem"], + }, + example_input={"problem": "无法登录"}, + ) + AgentStepBinding.objects.create( + agent_version=version, + code="structure", + order=1, + skill_version=first_skill, + input_mapping={"value": "$input.problem"}, + ) + if second_skill: + AgentStepBinding.objects.create( + agent_version=version, + code="reply", + order=2, + skill_version=second_skill, + input_mapping={"value": "$steps.structure.output.summary"}, + ) + if published: + definition.active_version = version + definition.save(update_fields=["active_version"]) + return definition, version diff --git a/backend/apps/agents/tests/test_agents_api.py b/backend/apps/agents/tests/test_agents_api.py new file mode 100644 index 0000000..0d7b995 --- /dev/null +++ b/backend/apps/agents/tests/test_agents_api.py @@ -0,0 +1,76 @@ +from django.contrib.auth import get_user_model +from django.core.management import call_command + +from apps.modelhub.services import CompletionResult + +from .helpers import create_agent, create_deployment + + +def login_admin(client): + call_command("bootstrap_governance") + user = get_user_model().objects.get(username="admin") + client.force_login(user) + return user + + +def test_admin_publishes_and_runs_agent(client, db, monkeypatch): + login_admin(client) + definition, version = create_agent() + monkeypatch.setattr( + "apps.modelhub.services._request_completion", + lambda *args, **kwargs: CompletionResult('{"summary":"ok","reply":"ok"}', 3, 2), + ) + + published = client.post(f"/api/agents/{definition.id}/publish") + assert published.status_code == 200 + assert published.json()["status"] == "published" + + run = client.post( + f"/api/agents/{definition.id}/runs", + data={"version": 1, "input": {"problem": "无法登录"}}, + content_type="application/json", + ) + assert run.status_code == 200 + assert run.json()["status"] == "succeeded" + public_id = run.json()["public_id"] + detail = client.get(f"/api/agent-runs/{public_id}") + assert detail.status_code == 200 + assert detail.json()["config_snapshot"]["agent"]["version"] == 1 + + +def test_publishing_rejects_draft_skill(client, db): + login_admin(client) + definition, version = create_agent() + binding = version.steps.first() + binding.skill_version.status = "draft" + binding.skill_version.save(update_fields=["status"]) + + response = client.post(f"/api/agents/{definition.id}/publish") + + assert response.status_code == 400 + assert "已发布 Skill" in response.json()["detail"] + + +def test_agent_run_permission_is_required(client, db): + definition, _ = create_agent(published=True) + user = get_user_model().objects.create_user(username="requester", password="password") + client.force_login(user) + + response = client.post( + f"/api/agents/{definition.id}/runs", + data={"version": 1, "input": {"problem": "无法登录"}}, + content_type="application/json", + ) + + assert response.status_code == 403 + + +def test_seed_command_creates_support_agent(client, db): + deployment = create_deployment() + + call_command("seed_ai_demo", deployment_id=deployment.id) + + from apps.agents.models import AgentDefinition + + agent = AgentDefinition.objects.get(code="support-analysis") + assert agent.versions.get(status="draft").steps.count() == 3 diff --git a/backend/apps/agents/tests/test_runtime.py b/backend/apps/agents/tests/test_runtime.py new file mode 100644 index 0000000..6ac01d3 --- /dev/null +++ b/backend/apps/agents/tests/test_runtime.py @@ -0,0 +1,114 @@ +from datetime import timedelta + +from django.utils import timezone + +from apps.agents.models import AgentRun +from apps.agents.services import build_config_snapshot, execute_agent_run +from apps.agents.tasks import cleanup_expired_agent_runs +from apps.modelhub.services import CompletionResult + +from .helpers import create_agent + + +def create_run(version): + return AgentRun.objects.create( + definition=version.definition, + agent_version=version, + sandbox=version.status == "draft", + input_payload={"problem": "无法登录"}, + config_snapshot=build_config_snapshot(version), + total_steps=version.steps.count(), + ) + + +def test_ordered_runtime_records_steps_usage_and_final_output(db, monkeypatch): + _, version = create_agent() + outputs = iter( + [ + CompletionResult('{"summary":"账号异常"}', 10, 5), + CompletionResult('{"reply":"请重置密码"}', 12, 6), + ] + ) + monkeypatch.setattr( + "apps.modelhub.services._request_completion", + lambda *args, **kwargs: next(outputs), + ) + run = create_run(version) + + execute_agent_run(run.id) + + run.refresh_from_db() + assert run.status == "succeeded" + assert run.final_output == {"reply": "请重置密码"} + assert run.input_tokens == 22 + assert run.output_tokens == 11 + assert list(run.steps.values_list("code", flat=True)) == ["structure", "reply"] + + +def test_invalid_structure_is_repaired_once(db, monkeypatch): + _, version = create_agent(two_steps=False) + outputs = iter( + [ + CompletionResult("not-json", 4, 2), + CompletionResult('{"summary":"已修复"}', 5, 3), + ] + ) + monkeypatch.setattr( + "apps.modelhub.services._request_completion", + lambda *args, **kwargs: next(outputs), + ) + run = create_run(version) + + execute_agent_run(run.id) + + run.refresh_from_db() + step = run.steps.get() + assert run.status == "succeeded" + assert step.attempts.count() == 2 + assert list(step.attempts.values_list("attempt_type", flat=True)) == ["initial", "repair"] + + +def test_second_invalid_structure_fails_and_stops(db, monkeypatch): + _, version = create_agent() + outputs = iter( + [ + CompletionResult("not-json", 4, 2), + CompletionResult("still-not-json", 5, 3), + ] + ) + monkeypatch.setattr( + "apps.modelhub.services._request_completion", + lambda *args, **kwargs: next(outputs), + ) + run = create_run(version) + + execute_agent_run(run.id) + + run.refresh_from_db() + assert run.status == "failed" + assert run.steps.count() == 1 + assert run.steps.get().status == "failed" + + +def test_cancellation_before_first_step_stops_run(db): + _, version = create_agent() + run = create_run(version) + run.cancellation_requested_at = timezone.now() + run.save(update_fields=["cancellation_requested_at"]) + + execute_agent_run(run.id) + + run.refresh_from_db() + assert run.status == "cancelled" + assert run.steps.count() == 0 + + +def test_cleanup_removes_expired_runs_but_keeps_audit(db, settings): + _, version = create_agent() + run = create_run(version) + AgentRun.objects.filter(id=run.id).update(created_at=timezone.now() - timedelta(days=91)) + + deleted = cleanup_expired_agent_runs() + + assert deleted == 1 + assert not AgentRun.objects.filter(id=run.id).exists() diff --git a/backend/config/settings/base.py b/backend/config/settings/base.py index fcc29f8..323f54e 100644 --- a/backend/config/settings/base.py +++ b/backend/config/settings/base.py @@ -23,6 +23,7 @@ INSTALLED_APPS = [ "apps.knowledge", "apps.modelhub", "apps.skills", + "apps.agents", ] MIDDLEWARE = [ diff --git a/backend/config/urls.py b/backend/config/urls.py index a9ca9a7..7d3b472 100644 --- a/backend/config/urls.py +++ b/backend/config/urls.py @@ -3,6 +3,8 @@ from django.urls import path from ninja import NinjaAPI from apps.accounts.api import router as accounts_router +from apps.agents.api import router as agents_router +from apps.agents.api import runs_router as agent_runs_router from apps.audit.api import router as audit_router from apps.knowledge.api import router as knowledge_router from apps.modelhub.api import router as modelhub_router @@ -20,6 +22,8 @@ api.add_router("/solutions", solutions_router) api.add_router("/knowledge", knowledge_router) api.add_router("/modelhub", modelhub_router) api.add_router("/skills", skills_router) +api.add_router("/agents", agents_router) +api.add_router("/agent-runs", agent_runs_router) urlpatterns = [ path("admin/", admin.site.urls), diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index 80429db..6a55bd8 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -62,6 +62,22 @@ services: volumes: - ../backend:/app + beat: + build: + context: .. + dockerfile: infra/backend.Dockerfile + command: celery -A config.celery beat -l info + env_file: + - path: ../.env.example + required: true + - path: ../.env + required: false + depends_on: + - backend + - redis + volumes: + - ../backend:/app + frontend: build: context: ..