智能体全流程的工程约束：从“能跑通”到“能交付”的结构性缺口

COMMANDER_PROMPT = """你是一名资深指挥官。请将用户需求拆解为3-5个逻辑步。
每个步骤必须包含以下字段：
{
  "step_id": "唯一标识",
  "description": "任务描述",
  "dependencies": ["前置步骤ID"],
  "expected_output": "预期输出格式",
  "assigned_expert": "建议的执行角色"
}
禁止将工具调用细节写入规划。规划只描述'做什么'，不描述'怎么做'。"""

from a2a.types import AgentCard, AgentSkill

agent_card = AgentCard(
    name="Deployment Agent",
    description="负责执行模型部署与回滚",
    version="1.0.0",
    skills=[
        AgentSkill(
            id="deploy_model",
            name="Deploy Model",
            description="将验证通过的模型部署到目标环境",
            input_schema={
                "model_id": "string",
                "environment": "staging|production",
                "idempotency_key": "string"
            },
            output_schema={
                "deployment_id": "string",
                "status": "succeeded|failed",
                "rollback_token": "string"
            }
        )
    ],
    url="https://deploy-agent.internal/a2a/jsonrpc"
)

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("k8s-ops-tools")

@mcp.tool()
def scale_deployment(service: str, replicas: int) -> str:
    """将 Kubernetes Deployment 缩放到指定副本数。

    Args:
        service: Deployment 名称
        replicas: 目标副本数，必须为正整数
    """
    if replicas <= 0:
        raise ValueError("副本数必须为正整数")
    import subprocess
    result = subprocess.run(
        ["kubectl", "scale", "deployment", service, f"--replicas={replicas}"],
        capture_output=True, text=True
    )
    return f"Scaled {service} to {replicas}: {result.stdout.strip()}"

class ContextManagedAgent:
    def __init__(self, llm, external_store):
        self.llm = llm
        self.store = external_store
        self.working_context = []

    def manage_context(self, threshold_ratio=0.85):
        """Agent自主判断是否压缩，而非固定token阈值触发"""
        if self._token_ratio() < threshold_ratio:
            return
        segment_id = self.store.write(self.working_context)
        summary = self.llm.summarize(
            self.working_context,
            system="保留所有实体名称、数值约束和未完成子目标。"
                   "标注原始记录的segment_id以便回查。"
        )
        self.working_context = [summary]

    def query_memory(self, segment_id, question):
        raw = self.store.read(segment_id)
        return self.llm.extract(raw, question)

class ReliabilityMonitor:
    def __init__(self, goals: list[dict]):
        self.goals = goals  # [{"goal_id": "g1", "criterion": "..."}]

    def check(self, step_result: dict, step_id: str) -> dict:
        """检查步骤结果是否满足其关联的目标标准"""
        related_goals = [
            g for g in self.goals
            if step_id in g.get("step_ids", [])
        ]
        violations = []
        for goal in related_goals:
            if not self._satisfies(step_result, goal["criterion"]):
                violations.append({
                    "goal_id": goal["goal_id"],
                    "criterion": goal["criterion"],
                    "actual": step_result.get("output")
                })
        return {
            "passed": len(violations) == 0,
            "violations": violations,
            "action": "proceed" if not violations else "replan"
        }

    def _satisfies(self, result, criterion):
        # 可接入规则引擎、LLM判断或人工审核队列
        return criterion.get("validator")(result)