527 lines
20 KiB
Python
527 lines
20 KiB
Python
import json
|
|
import time
|
|
|
|
from celery.result import AsyncResult
|
|
from django.conf import settings
|
|
from django.db import IntegrityError, transaction
|
|
from django.http import StreamingHttpResponse
|
|
from django.utils import timezone
|
|
from ninja import Router
|
|
from ninja.errors import HttpError
|
|
|
|
from apps.accounts.api import require_login, require_permission
|
|
from apps.agents.models import (
|
|
AgentDefinition,
|
|
AgentRun,
|
|
AgentStepBinding,
|
|
AgentVersion,
|
|
)
|
|
from apps.agents.schemas import (
|
|
AgentCreateIn,
|
|
AgentDefinitionOut,
|
|
AgentDefinitionUpdateIn,
|
|
AgentDetailOut,
|
|
AgentDraftUpdateIn,
|
|
AgentRunDetailOut,
|
|
AgentRunIn,
|
|
AgentRunListOut,
|
|
AgentVersionOut,
|
|
)
|
|
from apps.agents.services import (
|
|
AgentValidationError,
|
|
append_run_event,
|
|
ensure_agent_draft,
|
|
queue_agent_run,
|
|
replace_agent_steps,
|
|
validate_agent_version,
|
|
)
|
|
from apps.audit.models import OperationAuditLog
|
|
from apps.skills.services import SkillValidationError
|
|
|
|
router = Router(tags=["agents"])
|
|
runs_router = Router(tags=["agent-runs"])
|
|
|
|
|
|
def serialize_step(step: AgentStepBinding) -> dict:
|
|
deployment = step.model_deployment or step.agent_version.model_deployment
|
|
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,
|
|
"skill_output_enforcement": step.skill_version.output_enforcement,
|
|
"model_deployment_id": step.model_deployment_id,
|
|
"model_display_name": deployment.display_name,
|
|
"model_capability": deployment.capability,
|
|
"model_structured_output_mode": deployment.structured_output_mode,
|
|
"model_structured_output_test_status": deployment.structured_output_test_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,
|
|
"allowed_actions": version.allowed_actions,
|
|
"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,
|
|
"progress_stage": run.progress_stage,
|
|
"progress_message": run.progress_message,
|
|
"retry_count": run.retry_count,
|
|
"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,
|
|
"model_deployment_id": step.model_deployment_id,
|
|
"model_display_name": (
|
|
step.model_deployment.display_name
|
|
if step.model_deployment_id
|
|
else ""
|
|
),
|
|
"model_capability": (
|
|
step.model_deployment.capability
|
|
if step.model_deployment_id
|
|
else ""
|
|
),
|
|
"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": [
|
|
{
|
|
"id": attempt.id,
|
|
"attempt_type": attempt.attempt_type,
|
|
"status": attempt.status,
|
|
"raw_output": attempt.raw_output,
|
|
"validation_error": attempt.validation_error,
|
|
"input_tokens": attempt.input_tokens,
|
|
"output_tokens": attempt.output_tokens,
|
|
"duration_ms": attempt.duration_ms,
|
|
"structured_output": (
|
|
attempt.model_invocation.model_snapshot.get(
|
|
"structured_output"
|
|
)
|
|
or {}
|
|
if attempt.model_invocation_id
|
|
else {}
|
|
),
|
|
"created_at": attempt.created_at,
|
|
}
|
|
for attempt in step.attempts.select_related(
|
|
"model_invocation"
|
|
)
|
|
],
|
|
}
|
|
for step in run.steps.all()
|
|
],
|
|
}
|
|
)
|
|
return payload
|
|
|
|
|
|
def get_run_for_stream(request, public_id: str) -> AgentRun:
|
|
user = require_login(request)
|
|
run = AgentRun.objects.select_related("definition", "agent_version").filter(
|
|
public_id=public_id
|
|
).first()
|
|
if run is None:
|
|
raise HttpError(404, "Agent 运行不存在")
|
|
if (
|
|
run.created_by_id == user.id
|
|
or user.is_superuser
|
|
or user.has_perm("system.agent.run.view")
|
|
):
|
|
return run
|
|
raise HttpError(403, "Permission denied")
|
|
|
|
|
|
def parse_event_cursor(request, after: int | None) -> int:
|
|
header_value = request.headers.get("Last-Event-ID", "").strip()
|
|
try:
|
|
header_cursor = int(header_value) if header_value else 0
|
|
except ValueError as error:
|
|
raise HttpError(400, "Last-Event-ID 必须是非负整数") from error
|
|
query_cursor = after or 0
|
|
if header_cursor < 0 or query_cursor < 0:
|
|
raise HttpError(400, "事件游标必须是非负整数")
|
|
return max(header_cursor, query_cursor)
|
|
|
|
|
|
def format_sse_event(event) -> str:
|
|
data = json.dumps(
|
|
{
|
|
"sequence": event.sequence,
|
|
"event_type": event.event_type,
|
|
"payload": event.payload,
|
|
"created_at": event.created_at.isoformat(),
|
|
},
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
)
|
|
return f"id: {event.sequence}\nevent: {event.event_type}\ndata: {data}\n\n"
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
def validation_http_error(error: Exception) -> HttpError:
|
|
if isinstance(error, AgentValidationError) and error.code.startswith("structured_output_"):
|
|
return HttpError(400, f"{error.code}: {error}")
|
|
return HttpError(400, str(error))
|
|
|
|
|
|
@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,
|
|
allowed_actions=payload.allowed_actions,
|
|
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 validation_http_error(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 validation_http_error(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 validation_http_error(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 validation_http_error(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:
|
|
run = queue_agent_run(
|
|
version,
|
|
payload.input,
|
|
created_by=request.user,
|
|
sandbox=sandbox,
|
|
)
|
|
except (AgentValidationError, SkillValidationError) as error:
|
|
raise validation_http_error(error) from error
|
|
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=AgentRunListOut)
|
|
def list_agent_runs(
|
|
request,
|
|
status: str | None = None,
|
|
agent_id: int | None = None,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
):
|
|
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)
|
|
page = max(page, 1)
|
|
page_size = min(max(page_size, 1), 100)
|
|
total = queryset.count()
|
|
total_pages = (total + page_size - 1) // page_size if total else 0
|
|
start = (page - 1) * page_size
|
|
return {
|
|
"items": [serialize_run(run) for run in queryset[start : start + page_size]],
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total": total,
|
|
"total_pages": total_pages,
|
|
}
|
|
|
|
|
|
@runs_router.get("/{public_id}", response=AgentRunDetailOut)
|
|
def get_agent_run(request, public_id: str):
|
|
run = get_run_for_stream(request, public_id)
|
|
return serialize_run(run, detailed=True)
|
|
|
|
|
|
@runs_router.get("/{public_id}/events")
|
|
def stream_agent_run_events(request, public_id: str, after: int | None = None):
|
|
run = get_run_for_stream(request, public_id)
|
|
cursor = parse_event_cursor(request, after)
|
|
terminal_events = {"completed", "failed", "cancelled", "timed_out", "needs_input"}
|
|
max_seconds = getattr(settings, "AGENT_RUN_SSE_MAX_SECONDS", 60)
|
|
|
|
def event_stream():
|
|
current = cursor
|
|
started = time.monotonic()
|
|
last_heartbeat = started
|
|
while time.monotonic() - started < max_seconds:
|
|
events = list(run.events.filter(sequence__gt=current).order_by("sequence")[:100])
|
|
for event in events:
|
|
current = event.sequence
|
|
yield format_sse_event(event)
|
|
if events and events[-1].event_type in terminal_events:
|
|
return
|
|
latest = run.events.order_by("-sequence").first()
|
|
if (
|
|
latest is not None
|
|
and latest.sequence <= current
|
|
and latest.event_type in terminal_events
|
|
):
|
|
return
|
|
now = time.monotonic()
|
|
if now - last_heartbeat >= 15:
|
|
yield ": keep-alive\n\n"
|
|
last_heartbeat = now
|
|
time.sleep(0.5)
|
|
|
|
response = StreamingHttpResponse(event_stream(), content_type="text/event-stream")
|
|
response["Cache-Control"] = "no-cache"
|
|
response["X-Accel-Buffering"] = "no"
|
|
return response
|
|
|
|
|
|
@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.progress_stage = "cancelled"
|
|
run.progress_message = "运行已取消"
|
|
run.finished_at = timezone.now()
|
|
update_fields.extend(["status", "progress_stage", "progress_message", "finished_at"])
|
|
run.save(update_fields=update_fields)
|
|
if run.status == "cancelled":
|
|
append_run_event(run, "cancelled", {"message": "运行已取消"})
|
|
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)
|