"""Agent 版本校验、运行快照和多步骤执行编排。 运行时只使用已冻结的版本配置，确保模型、Skill、输入映射和授权动作可以事后还原。 """ import json import re from copy import deepcopy from time import monotonic from typing import Any from django.conf import settings from django.db import transaction from django.db.models import Max from django.utils import timezone from apps.agents.models import ( AgentDefinition, AgentRun, AgentRunAttempt, AgentRunEvent, 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 @transaction.atomic def append_run_event( run: AgentRun | int, event_type: str, payload: dict | None = None, ) -> AgentRunEvent: run_id = run.id if isinstance(run, AgentRun) else run locked_run = AgentRun.objects.select_for_update().get(id=run_id) if event_type in {"completed", "failed", "cancelled", "timed_out"}: existing = locked_run.events.order_by("-sequence").first() if existing is not None and existing.event_type == event_type: return existing last_sequence = locked_run.events.aggregate(value=Max("sequence"))["value"] or 0 event_payload = { "run_status": locked_run.status, "progress_stage": locked_run.progress_stage, "progress_message": locked_run.progress_message, **(payload or {}), } return AgentRunEvent.objects.create( run=locked_run, sequence=last_sequence + 1, event_type=event_type, payload=event_payload, ) def _set_run_progress(run: AgentRun, stage: str, message: str) -> None: run.progress_stage = stage run.progress_message = message run.save(update_fields=["progress_stage", "progress_message"]) event_type = stage if stage == "completed" and run.input_payload.get("draft_public_id"): event_type = "extraction_completed" append_run_event( run, event_type, {"current_step": run.current_step, "message": message}, ) def _step_progress(code: str, name: str) -> tuple[str, str]: if code == "facts": return "extracting", "正在解析沟通记录" if code == "draft": return "generating", "正在生成工单草稿" return "executing", f"正在执行：{name}" def validate_agent_version(version: AgentVersion, *, publishing: bool = False) -> None: """校验步骤顺序、模型能力、Skill 状态、输入映射和授权动作。""" validate_json_schema(version.input_schema, "Agent 输入 Schema") if version.example_input: validate_payload(version.input_schema, version.example_input, "Agent 示例输入") if ( not isinstance(version.allowed_actions, list) or any( not isinstance(action, str) or not action.strip() for action in version.allowed_actions ) or len(set(version.allowed_actions)) != len(version.allowed_actions) ): raise AgentValidationError("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", "model_deployment__provider", "agent_version__model_deployment__provider", ) ) if not steps: raise AgentValidationError("Agent 至少需要一个 Skill 步骤") if len(steps) > 10: raise AgentValidationError("Agent 最多允许 10 个步骤") known_steps: set[str] = set() for step in steps: deployment = step.model_deployment or version.model_deployment if deployment.capability not in {"text_chat", "vision_chat"}: raise AgentValidationError( f"步骤 {step.code} 必须绑定文本或视觉模型，当前为 {deployment.capability}" ) if publishing and (not deployment.enabled or not deployment.provider.enabled): raise AgentValidationError(f"步骤 {step.code} 绑定的模型部署未启用") 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), allowed_actions=deepcopy(source.allowed_actions), 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, model_deployment=step.model_deployment, input_mapping=deepcopy(step.input_mapping), ) for step in source.steps.all() ] ) return draft def build_config_snapshot(version: AgentVersion) -> dict: """冻结一次运行所需的 Agent、模型、Skill、映射和策略配置。""" deployment = version.model_deployment return { "agent": { "id": version.definition_id, "code": version.definition.code, "version": version.version, "status": version.status, "allowed_actions": deepcopy(version.allowed_actions), }, "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, "model": { "deployment_id": ( step.model_deployment_id or version.model_deployment_id ), "inherited": step.model_deployment_id is None, "provider_name": ( step.model_deployment.provider.name if step.model_deployment_id else deployment.provider.name ), "model_name": ( step.model_deployment.model_name if step.model_deployment_id else deployment.model_name ), "capability": ( step.model_deployment.capability if step.model_deployment_id else deployment.capability ), }, } for step in version.steps.select_related( "skill_version__definition", "model_deployment__provider", ) ], } def queue_agent_run( version: AgentVersion, input_payload: dict, *, created_by=None, sandbox: bool = False, config_snapshot_extra: dict | None = None, defer_until_commit: bool = False, ) -> AgentRun: """创建运行及初始事件，并按事务边界决定何时投递 Celery 任务。""" validate_agent_version(version, publishing=not sandbox) validate_payload(version.input_schema, input_payload, "Agent 输入") config_snapshot = build_config_snapshot(version) if config_snapshot_extra: config_snapshot.update(deepcopy(config_snapshot_extra)) run = AgentRun.objects.create( definition=version.definition, agent_version=version, sandbox=sandbox, input_payload=deepcopy(input_payload), config_snapshot=config_snapshot, total_steps=version.steps.count(), created_by=created_by, ) append_run_event(run, "queued", {"message": "Agent 运行已进入处理队列"}) def dispatch() -> None: from apps.agents.tasks import execute_agent_run try: result = execute_agent_run.delay(run.id) except Exception as error: run.status = "failed" run.progress_stage = "failed" run.progress_message = "Agent 运行入队失败" run.error_code = "queue_failed" run.error_message = str(error) run.finished_at = timezone.now() run.save( update_fields=[ "status", "progress_stage", "progress_message", "error_code", "error_message", "finished_at", ] ) append_run_event( run, "failed", {"error_code": run.error_code, "error_message": run.error_message}, ) return run.task_id = result.id or "" run.save(update_fields=["task_id"]) if defer_until_commit: transaction.on_commit(dispatch) else: dispatch() run.refresh_from_db() return run 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 _empty_schema_value(schema: dict) -> Any: value_type = schema.get("type") if value_type == "string" and (not schema.get("enum") or "" in schema["enum"]): return "" if value_type == "array": return [] if value_type == "object": return {} return None def _complete_object_shape(value: dict, schema: dict) -> dict: completed = dict(value) properties = schema.get("properties", {}) for code in schema.get("required", []): if code in completed: continue empty_value = _empty_schema_value(properties.get(code, {})) if empty_value is not None: completed[code] = empty_value return completed def _normalize_nested_output(output: Any, output_schema: dict) -> Any: """Repair common flattened model output without another model round-trip.""" if not isinstance(output, dict): return output properties = output_schema.get("properties", {}) required = set(output_schema.get("required", [])) if {"facts", "evidence", "missing_fields"}.issubset(required): fact_schema = properties.get("facts", {}) fact_codes = set(fact_schema.get("properties", {})) nested_facts = output.get("facts") if isinstance(nested_facts, dict): facts = _complete_object_shape(nested_facts, fact_schema) elif fact_codes.intersection(output): facts = _complete_object_shape( {code: output[code] for code in fact_codes if code in output}, fact_schema, ) else: facts = None if facts is not None: evidence = _complete_object_shape( output.get("evidence", {}) if isinstance(output.get("evidence"), dict) else {}, properties.get("evidence", {}), ) missing = list(output.get("missing_fields", [])) for code, value in facts.items(): if not value and code not in missing: missing.append(code) return {"facts": facts, "evidence": evidence, "missing_fields": missing} if {"fields", "missing_fields", "questions"}.issubset(required): field_schema = properties.get("fields", {}) field_codes = set(field_schema.get("properties", {})) nested_fields = output.get("fields") if isinstance(nested_fields, dict): fields = _complete_object_shape(nested_fields, field_schema) normalized = dict(output) elif field_codes.intersection(output): fields = _complete_object_shape( {code: output[code] for code in field_codes if code in output}, field_schema, ) normalized = { code: value for code, value in output.items() if code not in field_codes } else: fields = None if fields is not None: questions = output.get("questions", output.get("follow_up_questions", [])) normalized.pop("follow_up_questions", None) normalized.update( { "fields": fields, "missing_fields": list(output.get("missing_fields", [])), "questions": list(questions) if isinstance(questions, list) else [], } ) return normalized return output 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, image_parts: list[dict] | None = None, ) -> ModelInvocation: version = run.agent_version deployment = step.model_deployment or version.model_deployment messages = [] if version.system_prompt: messages.append({"role": "system", "content": version.system_prompt}) user_content: str | list[dict] = prompt step_image_parts = None if image_parts and deployment.capability == "vision_chat": if not deployment.provider.image_egress_approved: raise AgentValidationError( f"步骤 {step.code} 的模型渠道未授权图片出境" ) step_image_parts = image_parts if step_image_parts: user_content = [{"type": "text", "text": prompt}, *step_image_parts] messages.append({"role": "user", "content": user_content}) snapshot_messages = [ {"role": item["role"], "content": prompt if item["role"] == "user" else item["content"]} for item in messages ] invocation = ModelInvocation.objects.create( deployment=deployment, invocation_type="agent_step", input_snapshot={ "messages": snapshot_messages, "image_refs": ( run.input_payload.get("image_refs", []) if step_image_parts else [] ), "run_id": str(run.public_id), "step": step.code, }, created_by=run.created_by, ) if attempt_type == "structure_repair": _set_run_progress(run, "repairing", "正在修正结构化结果") def mark_retry(retry_number: int, _error_code: str) -> None: run.retry_count += 1 run.progress_stage = "retrying" run.progress_message = f"模型响应暂时不稳定，正在进行第 {retry_number} 次重试" run.save(update_fields=["retry_count", "progress_stage", "progress_message"]) execute_invocation( invocation, messages, temperature=version.temperature_override, max_output_tokens=version.max_output_tokens_override, on_retry=mark_retry, total_timeout_seconds=settings.AI_AGENT_MODEL_TIMEOUT_SECONDS, ) if invocation.status == "succeeded": _set_run_progress(run, "validating", "正在校验结构化结果") _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, "structure_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, *, image_parts: list[dict] | None = None) -> 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.progress_stage = "starting" run.progress_message = "正在准备 Agent 运行" run.save(update_fields=["status", "started_at", "progress_stage", "progress_message"]) append_run_event(run, "starting", {"message": run.progress_message}) step_outputs: dict[str, dict] = {} try: bindings = list( run.agent_version.steps.select_related( "skill_version__definition", "model_deployment__provider", "agent_version__model_deployment__provider", ) ) for binding in bindings: run.refresh_from_db(fields=["cancellation_requested_at"]) if run.cancellation_requested_at: run.status = "cancelled" _set_run_progress(run, "cancelled", "运行已取消") break run.current_step = binding.code run.save(update_fields=["current_step"]) progress_stage, progress_message = _step_progress( binding.code, binding.skill_version.definition.name, ) _set_run_progress(run, progress_stage, progress_message) step = AgentRunStep.objects.create( run=run, code=binding.code, name=binding.skill_version.definition.name, order=binding.order, skill_version=binding.skill_version, model_deployment=( binding.model_deployment or run.agent_version.model_deployment ), 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", image_parts) if invocation.status != "succeeded": raise AgentValidationError(invocation.error_message or "模型调用失败") try: output = _parse_json_output(invocation.output_text) output = _normalize_nested_output( output, binding.skill_version.output_schema, ) 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) output = _normalize_nested_output( output, binding.skill_version.output_schema, ) 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}" _set_run_progress(run, "failed", "Agent 运行失败") 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" _set_run_progress(run, "cancelled", "运行已取消") break if run.status == "running": run.final_output = step.output_payload if bindings else None try: from apps.tickets.support_analysis import validate_support_analysis_output validate_support_analysis_output(run, run.final_output) from apps.knowledge.distillation import ( validate_knowledge_distillation_output, ) validate_knowledge_distillation_output(run, run.final_output) except ValueError as error: run.status = "failed" run.error_code = ( "untrusted_support_output" if run.definition.code == "support-analysis" else "untrusted_knowledge_distillation_output" ) run.error_message = str(error) _set_run_progress(run, "failed", "Agent 结果未通过业务信任校验") else: run.status = "succeeded" _set_run_progress(run, "completed", "草稿生成完成") except Exception as error: run.status = "failed" run.error_code = "runtime_error" run.error_message = "Agent 运行发生未预期错误" _set_run_progress(run, "failed", "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() if run.definition.code == "support-analysis": from apps.workitems.services import sync_support_analysis_work_item if run.status == "succeeded": try: from apps.tickets.support_analysis import process_support_analysis_result process_support_analysis_result(run) except Exception as error: OperationAuditLog.objects.create( user=run.created_by, module="tickets", action="support_post_processing_failed", target_type="agent_run", target_id=str(run.public_id), request_snapshot={"error_type": type(error).__name__}, result="failed", ) sync_support_analysis_work_item( run, error_code="post_processing_failed", error_message="支持分析结果后处理失败，请人工接管", ) else: sync_support_analysis_work_item(run) else: sync_support_analysis_work_item(run) if run.definition.code == "knowledge-distillation": from apps.knowledge.models import TicketKnowledgeDistillation from apps.workitems.services import sync_knowledge_distillation_work_item try: distillation = run.ticket_knowledge_distillation except TicketKnowledgeDistillation.DoesNotExist: distillation = None if distillation is not None: if run.status == "succeeded": try: from apps.knowledge.distillation import ( process_knowledge_distillation_result, ) process_knowledge_distillation_result(run) except Exception as error: OperationAuditLog.objects.create( user=run.created_by, module="knowledge", action="knowledge_distillation_post_processing_failed", target_type="agent_run", target_id=str(run.public_id), request_snapshot={"error_type": type(error).__name__}, result="failed", ) sync_knowledge_distillation_work_item( distillation.ticket, run=run, error_code="post_processing_failed", error_message="知识蒸馏结果后处理失败，请人工接管", ) else: sync_knowledge_distillation_work_item( distillation.ticket, run=run, ) else: sync_knowledge_distillation_work_item( distillation.ticket, run=run, ) 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"], model_deployment_id=item.get("model_deployment_id"), input_mapping=item.get("input_mapping", {}), ) for index, item in enumerate(steps) ] ) validate_agent_version(version)