跳到正文
DEV Community· Alain Airom (Ayrom)·· 4 小時前AI 評分46

CUGA Agent Harness：主權 AI 智能體的架構設計、演進、實作與運作機制

Architecting Sovereign AI Agents: The Evolution, Implementation, and Mechanics of the CUGA Agent Harness

AI 導讀

CUGA（Configurable Generalist Agent）是 IBM Research 開源的企業級 Agent harness，透過受控架構支援主權 AI 智能體自動化。它將 LLM 作為規劃與推理引擎，並在模型外執行意圖防護、工具審批和工作流程規則，提供執行收據及軌跡記錄供審計。CUGA 可部署於私有雲或隔離環境，並連接本地模型與工具。

正文

News on CUGA and CUGA-Agent for Sovereignty.

It has been a while since I’ve written about CUGA, but CUGA has had a lot of changes and advances and could be used in a context of sovereign agent implementation. As enterprises and sovereign technical organizations increasingly demand local control, zero data leakage, and uncompromising guardrails over AI-driven automation, the choice of agent harness becomes paramount. Enter CUGA (Configurable Generalist Agent) — IBM Research’s open-source enterprise agent harness — paired with the comprehensive architecture of the cuga-agent-harness reference implementation.

In this deep-dive blog post, we will explore how CUGA bridges the gap between chaotic LLM behavior and predictable, enterprise-grade, sovereign execution. We will examine its architectural patterns, review production code excerpts from the harness, and discuss how to operationalize it securely.

An Intrduction to CUGA from Official GitHub Repository

In short, CUGA is the enterprise agent harness for sovereign, policy-controlled automation.

CUGA turns business requests into actions across your applications, APIs, and documents. Deploy in your private cloud or prepare an air-gapped environment, enforce fine-grained execution policies, and use open models on infrastructure you control. Connect your tools, configure your workflows, and optimize inference for your workload.

The Moto: Automate enterprise work. Keep control. 


| Your priority            | What CUGA gives you                                          |
| ------------------------ | ------------------------------------------------------------ |
| **Sovereignty**          | Run the harness in your environment. Connect privately hosted models and tools. Preload supporting model artifacts for disconnected deployments. |
| **Execution policies**   | Block disallowed intents, require approval for selected tools, guide workflows with playbooks, and configure tool-level guards. |
| **Inference efficiency** | Choose open models, tune reasoning modes, shortlist relevant tools, set tool-call budgets, and inspect token usage with per-run receipts. |
| **Faster delivery**      | Connect MCP servers, OpenAPI services, and Python tools. Configure and test an agent in the UI before publishing a version. |
| **Recurring automation** | Use the separate events service to run agents on schedules, webhooks, and application events, with run history in Events Studio. |
| **Reusable experience**  | Load task-specific skills and use a configured Evolve integration to retrieve relevant guidance and user preferences. |

CUGA in the Context of Sovereign Agent Implementation

The race toward autonomous agents often runs into a brick wall of operational realities: prompt drifting, hallucinations leading to unintended API calls, lack of data provenance, and strict data residency requirements (sovereignty). Traditional wrapper frameworks treat the LLM as the core orchestrator, leaving security and control up to fragile system prompts.

CUGA flips this paradigm. It treats the LLM as a planning and reasoning engine while enclosing it within a strict, deterministically enforced harness. This separation of concerns is what makes CUGA uniquely suited for sovereign agent implementations:

  • Zero Data Exfiltration: By running the underlying models locally via llama.cpp or Ollama, or securely through enterprise-managed, cloud-hosted LLMs within private virtual networks, every prompt, embedding, and tool execution stays entirely within air-gapped or localized perimeter perimeters.
  • Harness-Enforced Policies: Guardrails (such as Intent Guards, Tool Approvals, and Playbooks) are evaluated and enforced outside the model's immediate token stream, ensuring compliance even if the model attempts to bypass safety directives.
  • Auditable Execution Trails: Every run returns structured run receipts and deterministic trajectory logs, making auditing straightforward for compliance-heavy industries

Core Architecture: How the CUGA Showcase & Harness Operate

The CUGA Showcase and agent harness (cuga-agent-harness) are built around a framework-agnostic Python core (cuga_showcase/) paired with an intuitive Streamlit UI or headless execution interface.

System Architecture Flow

The system is designed with graceful degradation in mind. If a local model server becomes unreachable or returns an error, the BackendManager automatically manages fallbacks (e.g., from Live or local inference servers back to deterministic offline simulation) without crashing the user session.

Technical Stack

  • llama.cpp
  • Ollama
  • or with sovereign ‘on-premise cloud’ based LLMs.

Data for the Implementation

The 'CRM' data used for this demo application was generated synthetically.

"""Synthetic CRM data and the in-memory store the demo tools operate on.

All data is fictional and local, so the showcase has no cloud dependency.
"""

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

# Account records. ``usage`` is monthly active seats for the last six months.
_ACCOUNTS: Dict[str, Dict[str, Any]] = {
    "A-100": {
        "id": "A-100", "name": "Acme Corp", "plan": "Enterprise", "arr": 240000,
        "renewal_date": "2026-12-01", "nps": 52, "owner": "Dana Whitfield",
        "usage": [820, 845, 890, 910, 960, 1010],
    },
    "A-200": {
        "id": "A-200", "name": "Globex", "plan": "Pro", "arr": 96000,
        "renewal_date": "2026-11-15", "nps": 18, "owner": "Marcus Lee",
        "usage": [700, 660, 610, 540, 500, 455],
    },
    "A-300": {
        "id": "A-300", "name": "Initech", "plan": "Starter", "arr": 18000,
        "renewal_date": "2027-02-01", "nps": 35, "owner": "Priya Nair",
        "usage": [200, 201, 199, 202, 200, 203],
    },
    "A-400": {
        "id": "A-400", "name": "Umbrella", "plan": "Enterprise", "arr": 410000,
        "renewal_date": "2026-12-20", "nps": 41, "owner": "Dana Whitfield",
        "usage": [1500, 1490, 1470, 1460, 1440, 1430],
    },
}

_TICKETS: List[Dict[str, Any]] = [
    {"id": "T-1", "account_id": "A-100", "priority": "P3", "title": "Dashboard colour palette request", "status": "open"},
    {"id": "T-2", "account_id": "A-200", "priority": "P1", "title": "API outage in EU region", "status": "open"},
    {"id": "T-3", "account_id": "A-200", "priority": "P2", "title": "Export job times out", "status": "open"},
    {"id": "T-4", "account_id": "A-200", "priority": "P2", "title": "SSO login loop", "status": "open"},
    {"id": "T-5", "account_id": "A-400", "priority": "P2", "title": "Slow report generation", "status": "open"},
    {"id": "T-6", "account_id": "A-300", "priority": "P3", "title": "Invoice address change", "status": "closed"},
]

# Health score penalties per open ticket priority.
_TICKET_PENALTY = {"P1": 15, "P2": 6, "P3": 2}


def usage_trend_pct(usage: List[int]) -> float:
    """Percent change of the last three months vs the three months before."""
    if len(usage) < 6:
        return 0.0
    prev = sum(usage[-6:-3]) / 3
    last = sum(usage[-3:]) / 3
    return round((last - prev) / prev * 100, 1) if prev else 0.0


def health_label(score: int) -> str:
    """Map a numeric score to a text label (never colour alone)."""
    if score >= 75:
        return "healthy"
    if score >= 50:
        return "needs attention"
    return "at risk"


def compute_health(account: Dict[str, Any], tickets: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Deterministic account health score in the 0-100 range.

    score = 100 - ticket penalties + trend bonus (capped +-15) + NPS adjustment
    """
    penalty = sum(_TICKET_PENALTY.get(t["priority"], 0) for t in tickets if t["status"] == "open")
    trend = usage_trend_pct(account["usage"])
    trend_bonus = max(-15.0, min(15.0, trend * 0.8))
    nps_adj = (account["nps"] - 30) * 0.2
    score = int(round(max(0.0, min(100.0, 100 - penalty + trend_bonus + nps_adj - 10))))
    return {
        "account_id": account["id"],
        "score": score,
        "label": health_label(score),
        "drivers": {
            "ticket_penalty": penalty,
            "usage_trend_pct": trend,
            "nps": account["nps"],
        },
    }


@dataclass
class CrmStore:
    """Mutable in-memory CRM. One instance per browser session."""

    accounts: Dict[str, Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_ACCOUNTS))
    tickets: List[Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_TICKETS))
    outbox: List[Dict[str, Any]] = field(default_factory=list)
    deleted: List[str] = field(default_factory=list)

    def find_account(self, query: str) -> Optional[Dict[str, Any]]:
        """Find an account by id or (partial, case-insensitive) name."""
        q = (query or "").strip().lower()
        if not q:
            return None
        for acc in self.accounts.values():
            if acc["id"].lower() == q or acc["name"].lower() == q:
                return acc
        for acc in self.accounts.values():
            if q in acc["name"].lower():
                return acc
        return None

    def open_tickets(self, account_id: str) -> List[Dict[str, Any]]:
        return [t for t in self.tickets if t["account_id"] == account_id and t["status"] == "open"]

    def reset(self) -> None:
        """Restore the pristine demo data."""
        self.accounts = copy.deepcopy(_ACCOUNTS)
        self.tickets = copy.deepcopy(_TICKETS)
        self.outbox.clear()
        self.deleted.clear()

Implementation: Architecture, Policies, and Code

To see how this works in practice, let's examine the core components that govern tool execution, policy enforcement, and runtime receipts within the harness.

  • Tool Envelopes and Safety Envelopes: every action an agent can take is wrapped in a standardized tool specification. To ensure the model can gracefully recover from execution errors (such as missing parameters or unauthorized states), tools return standardized JSON envelopes rather than throwing unhandled exceptions: 
{"ok": True,  "data": {...}}
{"ok": False, "code": "not_found", "error": "No account matches 'Foo'."}

The code below shows how tools are registered and categorized into distinct sub-agents under a supervisor framework (cuga_showcase/tools.py pattern):

"""Demo tools shared by the offline simulator and the live ``CugaAgent``.

Each tool is a plain function returning the *envelope* convention used by the
official cuga-apps reference apps::

    {"ok": True,  "data": ...}
    {"ok": False, "code": "not_found", "error": "..."}

``ToolSpec`` carries the metadata (description, sensitivity, owning sub-agent)
so the same registry can be exposed as LangChain tools in Live mode.
"""

from __future__ import annotations

import ast
import operator
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

from .demo_data import CrmStore, compute_health, usage_trend_pct

Envelope = Dict[str, Any]


def ok(data: Any) -> Envelope:
    """Success envelope."""
    return {"ok": True, "data": data}


def err(code: str, message: str) -> Envelope:
    """Failure envelope."""
    return {"ok": False, "code": code, "error": message}


# --------------------------------------------------------------------------
# Safe calculator (no eval): walks the AST and allows arithmetic only.
# --------------------------------------------------------------------------
_BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.Mod: operator.mod,
}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def _eval_node(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
        left, right = _eval_node(node.left), _eval_node(node.right)
        if isinstance(node.op, ast.Pow) and abs(right) > 12:
            raise ValueError("exponent too large")
        return _BIN_OPS[type(node.op)](left, right)
    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
        return _UNARY_OPS[type(node.op)](_eval_node(node.operand))
    raise ValueError("unsupported expression")


def calculator(expression: str) -> Envelope:
    """Evaluate an arithmetic expression such as ``(240000 + 96000) * 0.1``."""
    try:
        tree = ast.parse((expression or "").strip(), mode="eval")
        value = _eval_node(tree.body)
    except ZeroDivisionError:
        return err("math_error", "Division by zero.")
    except (ValueError, SyntaxError):
        return err("bad_expression", "Only plain arithmetic is supported.")
    return ok({"expression": expression, "value": round(value, 6)})


# --------------------------------------------------------------------------
# CRM tools (bound to a CrmStore through ``build_registry``)
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class ToolSpec:
    """Metadata + callable for one tool."""

    name: str
    description: str
    fn: Callable[..., Envelope]
    sensitive: bool = False  # sensitive tools are the usual approval candidates
    agent: str = "cuga"  # owning sub-agent in supervisor mode
    params: Optional[Dict[str, str]] = None  # param name -> short description

    def call(self, **kwargs: Any) -> Envelope:
        """Invoke the tool, converting unexpected exceptions into envelopes."""
        try:
            return self.fn(**kwargs)
        except TypeError as exc:  # wrong/missing arguments
            return err("bad_arguments", str(exc))
        except Exception as exc:  # noqa: BLE001 - tools must never crash a run
            return err("tool_error", str(exc))


def build_registry(
    store: CrmStore, knowledge_search: Optional[Callable[[str], Envelope]] = None
) -> Dict[str, ToolSpec]:
    """Create the tool registry bound to ``store``.

    Args:
        store: The session's CRM store.
        knowledge_search: Optional callable backing ``search_knowledge``.
    """

    def get_account(query: str) -> Envelope:
        acc = store.find_account(query)
        if not acc:
            return err("not_found", f"No account matches '{query}'.")
        return ok({k: v for k, v in acc.items() if k != "usage"})

    def list_open_tickets(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(store.open_tickets(acc["id"]))

    def compute_health_score(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(compute_health(acc, store.open_tickets(acc["id"])))

    def get_usage_trend(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok({"account_id": acc["id"], "monthly_seats": acc["usage"], "trend_pct": usage_trend_pct(acc["usage"])})

    def list_accounts() -> Envelope:
        return ok([{"id": a["id"], "name": a["name"], "plan": a["plan"], "arr": a["arr"]} for a in store.accounts.values()])

    def send_email(to: str, subject: str, body: str) -> Envelope:
        if "@" not in (to or ""):
            return err("bad_arguments", "Recipient must be an email address.")
        store.outbox.append({"to": to, "subject": subject, "body": body})
        return ok({"queued": True, "to": to, "subject": subject})

    def delete_account(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        store.accounts.pop(acc["id"])
        store.deleted.append(acc["id"])
        return ok({"deleted": acc["id"], "name": acc["name"]})

    def search_knowledge(query: str) -> Envelope:
        if knowledge_search is None:
            return err("unavailable", "Knowledge base is not configured.")
        return knowledge_search(query)

    specs = [
        ToolSpec("list_accounts", "List all CRM accounts.", list_accounts, agent="crm_agent", params={}),
        ToolSpec("get_account", "Look up one account by id or name.", get_account, agent="crm_agent",
                 params={"query": "Account id (A-100) or name (Acme)"}),
        ToolSpec("get_usage_trend", "Monthly active seats and 3-month trend for an account.", get_usage_trend,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
        ToolSpec("list_open_tickets", "Open support tickets for an account.", list_open_tickets, agent="support_agent",
                 params={"account_id": "Account id or name"}),
        ToolSpec("compute_health_score", "Deterministic 0-100 health score with drivers.", compute_health_score,
                 agent="support_agent", params={"account_id": "Account id or name"}),
        ToolSpec("calculator", "Evaluate an arithmetic expression.", calculator, agent="crm_agent",
                 params={"expression": "Arithmetic such as 240000*0.1"}),
        ToolSpec("search_knowledge", "Search the local knowledge base and return cited passages.", search_knowledge,
                 agent="crm_agent", params={"query": "Natural-language question"}),
        ToolSpec("send_email", "Queue an email to a contact.", send_email, sensitive=True, agent="comms_agent",
                 params={"to": "Recipient address", "subject": "Subject line", "body": "Message body"}),
        ToolSpec("delete_account", "Permanently delete an account.", delete_account, sensitive=True,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
    ]
    return {s.name: s for s in specs}


def tool_names(registry: Dict[str, ToolSpec]) -> List[str]:
    """Stable, sorted tool names."""
    return sorted(registry)
  • Enforcing Harness Policies: policies steer the agent outside the prompt. Five policy types are natively supported in the CUGA harness: **Intent Guard, Playbook, Tool Guide, Tool Approval, and Output Formatter.

When a sensitive tool like delete_account is invoked, a Tool Approval policy intercepts the execution flow, parks the run state (awaiting_approval), and requests human validation before letting the tool execute: 


"""Policy engine for the offline simulator.

It reproduces, in plain Python, the *behaviour* of CUGA's five policy types so
users can experiment without an LLM:

* ``intent_guard``     - blocks a request before any planning happens
* ``playbook``         - injects step-by-step guidance when a request matches
* ``tool_guide``       - enriches a tool's description with extra guidance
* ``tool_approval``    - parks a tool call until a human approves it (HITL)
* ``output_formatter`` - reshapes the final answer

In Live mode the very same definitions are registered on the real
``agent.policies`` manager (see ``live.py``).
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from typing import Dict, List, Optional

POLICY_TYPES = ("intent_guard", "playbook", "tool_guide", "tool_approval", "output_formatter")

TYPE_LABELS = {
    "intent_guard": "Intent Guard",
    "playbook": "Playbook",
    "tool_guide": "Tool Guide",
    "tool_approval": "Tool Approval",
    "output_formatter": "Output Formatter",
}

TYPE_HELP = {
    "intent_guard": "Blocks a request that matches keywords before the agent plans anything.",
    "playbook": "Adds step-by-step guidance when the request matches keywords.",
    "tool_guide": "Appends usage guidance to the descriptions of the target tools.",
    "tool_approval": "Pauses before the listed tools run and asks a human to approve.",
    "output_formatter": "Rewrites the final answer using a format instruction.",
}


@dataclass
class Policy:
    """One policy definition (superset of the fields of the five SDK types)."""

    name: str
    type: str
    keywords: List[str] = field(default_factory=list)
    content: str = ""  # playbook / tool_guide content, formatter instruction
    response: str = ""  # intent_guard response
    tools: List[str] = field(default_factory=list)  # tool_approval / tool_guide targets
    message: str = ""  # tool_approval prompt
    enabled: bool = True
    id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])

    def matches(self, text: str) -> bool:
        """Case-insensitive keyword OR-match (the SDK's default operator)."""
        lowered = (text or "").lower()
        return any(k.strip().lower() in lowered for k in self.keywords if k.strip())


def default_policies() -> List[Policy]:
    """Starter policy set shown in the Policy Lab."""
    return [
        Policy(
            name="Block salary disclosure", type="intent_guard",
            keywords=["salary", "compensation of", "payroll"],
            response="I can't share compensation data. Please contact HR.",
        ),
        Policy(
            name="Churn-risk playbook", type="playbook",
            keywords=["churn", "at risk", "renewal risk"],
            content="1. Fetch the account.\n2. Check open P1/P2 tickets.\n3. Compare the 3-month usage trend.\n4. Recommend one concrete next action.",
        ),
        Policy(
            name="Email etiquette", type="tool_guide", tools=["send_email"],
            content="Keep emails under 120 words. Always include the account owner in the greeting.",
        ),
        Policy(
            name="Approve outbound & destructive actions", type="tool_approval",
            tools=["send_email", "delete_account"],
            message="This action changes data outside the conversation. Approve?",
        ),
        Policy(
            name="Executive summary format", type="output_formatter",
            keywords=[], content="Answer as a bold one-line verdict followed by up to three bullet points.",
            enabled=False,
        ),
    ]


class PolicySet:
    """A mutable collection of policies with the lookups the simulator needs."""

    def __init__(self, policies: Optional[List[Policy]] = None) -> None:
        self._items: Dict[str, Policy] = {}
        for p in policies if policies is not None else default_policies():
            self._items[p.id] = p

    # -- CRUD ---------------------------------------------------------------
    def add(self, policy: Policy) -> Policy:
        if policy.type not in POLICY_TYPES:
            raise ValueError(f"Unknown policy type: {policy.type}")
        if not policy.name.strip():
            raise ValueError("Policy name is required.")
        self._items[policy.id] = policy
        return policy

    def remove(self, policy_id: str) -> bool:
        return self._items.pop(policy_id, None) is not None

    def clear(self) -> None:
        self._items.clear()

    def toggle(self, policy_id: str, enabled: bool) -> None:
        if policy_id in self._items:
            self._items[policy_id].enabled = enabled

    def all(self) -> List[Policy]:
        return list(self._items.values())

    def active(self, ptype: str) -> List[Policy]:
        return [p for p in self._items.values() if p.type == ptype and p.enabled]

    # -- lookups used by the agent ------------------------------------------
    def intent_guard(self, message: str) -> Optional[Policy]:
        return next((p for p in self.active("intent_guard") if p.matches(message)), None)

    def playbooks(self, message: str) -> List[Policy]:
        return [p for p in self.active("playbook") if p.matches(message)]

    def tool_guides(self, tool: str) -> List[Policy]:
        return [p for p in self.active("tool_guide") if tool in p.tools or "*" in p.tools]

    def approval_for(self, tool: str) -> Optional[Policy]:
        return next((p for p in self.active("tool_approval") if tool in p.tools or "*" in p.tools), None)

    def formatter(self) -> Optional[Policy]:
        return next(iter(self.active("output_formatter")), None)
  • Dynamic Capability Injection via SKILL.md: a standout architectural feature of CUGA is its implementation of skills. Instead of cluttering the system prompt with hundreds of rigid instructions for every conceivable task, skills are modularized into independent directories containing a SKILL.md file featuring YAML front matter (name, description, tags) followed by procedural instructions. 
"""Loader for ``SKILL.md`` files (the format used by cuga-skills).

A skill is a Markdown file with YAML-style front matter (``name``,
``description``, ``tags``) followed by instructions. CUGA loads skills from
``.cuga/skills``; here we also match them to requests in the simulator.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

from .knowledge import tokenize


@dataclass(frozen=True)
class Skill:
    name: str
    description: str
    body: str
    tags: Tuple[str, ...] = field(default_factory=tuple)
    path: str = ""

    def score(self, message: str) -> float:
        """Share of the description's keywords found in ``message`` (0-1)."""
        wanted = set(tokenize(self.description))
        if not wanted:
            return 0.0
        return len(wanted & set(tokenize(message))) / len(wanted)


def parse_skill(text: str, path: str = "") -> Optional[Skill]:
    """Parse one SKILL.md; returns None when the front matter is invalid."""
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", text, re.DOTALL)
    if not match:
        return None
    meta: Dict[str, object] = {}
    current_list: Optional[str] = None
    for line in match.group(1).splitlines():
        if re.match(r"^\s+-\s+", line) and current_list:
            meta.setdefault(current_list, []).append(re.sub(r"^\s+-\s+", "", line).strip())  # type: ignore[union-attr]
            continue
        kv = re.match(r"^([A-Za-z_]+):\s*(.*)$", line)
        if kv:
            key, value = kv.group(1), kv.group(2).strip()
            if value:
                meta[key] = value
                current_list = None
            else:
                meta[key] = []
                current_list = key
    name, description = meta.get("name"), meta.get("description")
    if not isinstance(name, str) or not isinstance(description, str):
        return None
    tags = tuple(meta.get("tags", [])) if isinstance(meta.get("tags"), list) else ()
    return Skill(name=name, description=description, body=match.group(2).strip(), tags=tags, path=path)


def load_skills(folder: Path) -> List[Skill]:
    """Load every ``*/SKILL.md`` below ``folder`` (invalid files are skipped)."""
    skills: List[Skill] = []
    if not folder.is_dir():
        return skills
    for path in sorted(folder.rglob("SKILL.md")):
        try:
            skill = parse_skill(path.read_text(encoding="utf-8"), str(path))
        except (OSError, UnicodeDecodeError):
            continue
        if skill:
            skills.append(skill)
    return skills


def match_skill(skills: List[Skill], message: str, threshold: float = 0.25) -> Optional[Skill]:
    """Best skill for ``message`` when its score reaches ``threshold``."""
    best = max(skills, key=lambda s: s.score(message), default=None)
    return best if best and best.score(message) >= threshold else None

---
name: account-health-review
description: Use when the user asks how healthy, risky or satisfied a customer account is, or whether it may churn.
tags:
  - crm
---

# Account health review

1. Look up the account with `get_account`.
2. Call `compute_health_score` and `list_open_tickets`.
3. Call `get_usage_trend` and report the three-month trend.
4. Finish with one verdict (healthy, needs attention, at risk) and one recommended next action.

The current skills in the catalog:

  • incident-summary: Automatically invoked when support tickets need triage, grouping by priority (P1 first), and recommending the next remediation steps.

  • churn-risk-playbook: Evaluates active usage drops, open critical tickets, and ARR value to compile a comprehensive executive risk report.

This modular skill design ensures that sovereign deployments can easily drop in, update, or remove operational workflows without recompiling core agent codebases. The agent reads only the metadata descriptions and dynamically loads the full skill instruction set on demand when a user request requires it

Persona-Based Implementation & Visual Workflows

To make sovereign agent deployments practical, the cuga-agent-harness models real-world enterprise operations through concrete user stories and visual interactive artifacts (.bob/artifacts/user-stories-20261017.md and .bob/artifacts/user-stories-viz-20261017.html).

Persona 1: Sarah (Enterprise Customer Success Manager)

  • Goal: Quickly assess account health and proactively identify churn risks across enterprise accounts without manual CRM mining.

  • CUGA Implementation: Sarah interacts with the Supervisor agent, asking: "Which accounts are most at risk?" The CugaSupervisor delegates tasks to the crm_agent and support_agent, compiling usage trends, support ticket counts, and ARR data into a unified, cited report.

Persona 2: Marcus (SecOps & Compliance Officer)

  • Goal: Ensure all autonomous AI operations comply with strict corporate security policies and data residency mandates.

  • CUGA Implementation: Marcus configures Tool Approval policies for high-consequence tools
    (such as delete_account or send_email). When an agent attempts to purge a record, the harness intercepts the call, transitions the state to awaiting_approval, and halts execution until Marcus explicitly clicks Approve or Deny in the UI.

Persona 3: Elena (Local AI Developer / Infrastructure Engineer)

  • Goal: Deploy and run advanced agentic workflows locally on air-gapped workstations using local models.

  • CUGA Implementation: Elena utilizes the llama.cpp, Ollama local backends or cloud based LLMs with automatic health checks and graceful fallbacks, inspecting detailed Run Receipts to verify token consumption, wall-clock timing, and tool execution latencies.

Visualizing the Application and Use-Cases

The interactive visual artifact (https://github.com/aairom/cuga-agent-harness/blob/main/.bob/artifacts/user-stories-viz-20261017.html) provides a rendered, node-based representation of these persona pathways, mapping user intents directly to harness policy validation gates, multi-agent delegations, and tool envelope outcomes.

Inspecting Run Receipts: unlike standard LLM telemetry that tracks monetary costs (which vary wildly and complicate sovereign budgeting), CUGA's run receipt focuses strictly on cryptographic and operational determinism: tokens, LLM call counts, wall-clock timing, and tool execution durations.

Conclusion

The evolution of CUGA as demonstrated in the cuga-agent-harness proves that autonomous AI agents do not need to operate as black boxes. By combining local LLM backends (via Ollama or llama.cpp), or cloud based LLM connections, deterministic harness-level policy enforcement, and structured tool envelopes, organizations can finally deploy sovereign agents that respect data boundaries, require human-in-the-loop approvals for critical actions, and provide rigorous audit trails.

Whether you are modernizing legacy enterprise workflows or building specialized local development assistants, CUGA offers a robust blueprint for trustworthy, sovereign agent architecture.

Thanks for reading🦉

Links and references

來源：DEV Community · dev.to

#教學/實踐#Agent#部署/工程#開源生態