Add Reusable Templates to an AI Assistant with AgentScope 2.0.3

Overview

AgentScope 2.0.3 introduces a Skills layer that extracts business capabilities into reusable, hot‑updatable, governed templates. This separates ability description from execution platform, enabling zero‑code integration for scenarios such as order queries, refunds, and marketing recommendations.

Why Skills Instead of a Simple Prompt+Tool Stack

Business logic is tightly coupled to code when a single prompt and a flat tool list are used.

Any change in business rules forces a code release, slowing iteration.

All requests share the same LLM call chain, causing latency spikes and higher failure rates under load.

Session state, retry, rate‑limit, audit, gray‑release, and rollback are handled by ad‑hoc patches.

Skills separate the ability description from the execution platform>, turning the assistant into a composable, governable service.

Real Business Scenario: High‑Traffic Customer Service

Order query – read‑only, high‑frequency, low risk.

After‑sale processing – write‑heavy, medium risk.

Marketing recommendation – compute‑heavy, high token cost.

These intents require completely different execution chains; a monolithic prompt mixes them, leading to capability, release, resource, and governance coupling.

Four Core Problems of the Old Approach

Capability coupling – all domains share the same prompt and tool set.

Release coupling – any new business rule forces a new deployment.

Resource coupling – high‑priority write traffic competes with low‑priority recommendation traffic.

Governance coupling – session memory, idempotency, audit, retry, and rate‑limit are scattered in business code.

Correct Architecture with AgentScope 2.0.3

The system is split into three layers:

Business Template Layer (Skills) : Declarative, versioned, gray‑releaseable ability description.

Runtime Control Layer (Skill Runtime / Session / Guardrails) : Session management, tool authentication, idempotency, timeout, retry, circuit‑breaker, audit, hot‑reload, manual takeover.

Infrastructure Layer : API gateway, Kafka, Redis, MySQL, OpenTelemetry, Kubernetes.

Business Template Layer (Skills)

A Skill answers five questions: what the ability does, its input/output schema, allowed tools, execution flow, and version/gray‑release policy. The Skill is not a prompt; it is a governed capability package.

Prompt solves "what the model should say"
Skill solves "how the ability is described, controlled, and versioned"

Example refund_request.yaml (simplified):

name: refund_request
version: 2.1.0
description: Process user refund requests, verify order, evaluate risk, create ticket
labels:
  domain: customer-service
  priority: high
  write_operation: true
input_schema:
  type: object
  properties:
    session_id: {type: string}
    tenant_id: {type: string}
    user_id: {type: string}
    order_id: {type: string}
    refund_reason: {type: string, minLength: 2, maxLength: 200}
  required: [session_id, tenant_id, user_id, order_id, refund_reason]
output_schema:
  type: object
  properties:
    accepted: {type: boolean}
    ticket_id: {type: string}
    message: {type: string}
    next_action: {type: string}
  required: [accepted, message]
tools:
  - name: get_order_detail
    type: http
    timeout_ms: 800
    retry:
      max_attempts: 2
    auth:
      mode: service_token
    endpoint: "${ORDER_SERVICE_URL}/api/orders/detail"
  - name: evaluate_refund_risk
    type: http
    timeout_ms: 500
    retry:
      max_attempts: 1
    endpoint: "${RISK_SERVICE_URL}/api/refund/evaluate"
  - name: create_refund_ticket
    type: http
    timeout_ms: 1200
    retry:
      max_attempts: 0
    idempotent: true
    endpoint: "${AFTERSALE_SERVICE_URL}/api/refund/tickets"
prompt_template: |
  你是电商平台的售后受理助手。
  你的职责不是自由聊天，而是按照业务规则处理退款申请。
  必须遵守:
  1. 仅在订单归属当前用户时继续处理；
  2. 风控高风险时不得创建退款工单；
  3. 所有结论必须基于工具返回结果，不得编造；
  4. 输出必须符合 output_schema。
flow:
  type: sequential
  steps:
    - action: tool
      tool: get_order_detail
      args:
        tenantId: "{{ input.tenant_id }}"
        userId: "{{ input.user_id }}"
        orderId: "{{ input.order_id }}"
    - action: condition
      expression: "{{ get_order_detail.result.ownerMatched == false }}"
      when_true:
        - action: respond
          data:
            accepted: false
            message: "订单与当前用户不匹配，无法受理退款申请。"
            next_action: "manual_check"
    - action: tool
      tool: evaluate_refund_risk
      args:
        tenantId: "{{ input.tenant_id }}"
        userId: "{{ input.user_id }}"
        orderId: "{{ input.order_id }}"
        reason: "{{ input.refund_reason }}"
    - action: condition
      expression: "{{ evaluate_refund_risk.result.level == 'HIGH' }}"
      when_true:
        - action: respond
          data:
            accepted: false
            message: "系统检测到高风险退款申请，已转人工复核。"
            next_action: "manual_review"
    - action: tool
      tool: create_refund_ticket
      args:
        tenantId: "{{ input.tenant_id }}"
        userId: "{{ input.user_id }}"
        orderId: "{{ input.order_id }}"
        reason: "{{ input.refund_reason }}"
        requestId: "{{ context.request_id }}"
    - action: respond
      data:
        accepted: true
        ticket_id: "{{ create_refund_ticket.result.ticketId }}"
        message: "退款申请已提交，售后工单已创建。"
        next_action: "wait_audit"
memory_policy:
  short_term:
    enabled: true
    max_turns: 8
    summary_after_turns: 5
  long_term:
    enabled: false
error_policy:
  on_tool_timeout: fail_fast
  on_validation_error: reject_request
  on_unhandled_exception: fallback_to_human

The definition separates business flow (the flow steps) from platform concerns (session, retry, audit).

Runtime Control Layer

Session management

Tool authentication and whitelist

Idempotency control

Timeout, retry, circuit‑breaker

Audit logging

Hot reload & version switch

Manual takeover

Infrastructure Layer

API Gateway – auth, tenant isolation, rate‑limit.

Kafka – async decoupling and back‑pressure.

Redis – session cache, idempotency keys, hot‑skill cache.

MySQL – task state, tool audit, manual review records.

OpenTelemetry / Prometheus / Loki – tracing, metrics, logs.

Kubernetes – elastic scaling, gray‑release via deployment strategies.

Task‑Based Execution for High Concurrency

Read‑only, low‑latency queries can be handled synchronously. Write‑heavy or long‑chain scenarios are turned into tasks.

# FastAPI endpoint for creating a task
router = APIRouter(prefix="/api/agent/tasks", tags=["agent"])

class CreateTaskRequest(BaseModel):
    skill_name: str = Field(min_length=2, max_length=64)
    tenant_id: str = Field(min_length=1, max_length=64)
    user_id: str = Field(min_length=1, max_length=64)
    session_id: str = Field(min_length=1, max_length=64)
    payload: dict

class CreateTaskResponse(BaseModel):
    task_id: str
    accepted: bool
    status: str

@router.post("", response_model=CreateTaskResponse)
async def create_task(req: CreateTaskRequest, x_request_id: str | None = Header(default=None)):
    if req.skill_name not in {"order_query", "refund_request", "recommend_offer"}:
        raise HTTPException(status_code=400, detail="unsupported skill")
    task_id = await create_skill_task(
        request_id=x_request_id,
        skill_name=req.skill_name,
        tenant_id=req.tenant_id,
        user_id=req.user_id,
        session_id=req.session_id,
        payload=req.payload,
    )
    return CreateTaskResponse(task_id=task_id, accepted=True, status="PENDING")

The function performs deduplication via Redis SETNX, stores a task record in MySQL, and routes the task to a skill‑specific Kafka topic (e.g., skill-readonly-high, skill-write-critical, skill-recommend-low).

async def create_skill_task(request_id, skill_name, tenant_id, user_id, session_id, payload) -> str:
    task_id = str(uuid.uuid4())
    request_key = request_id or task_id
    dedup_key = f"skill:dedup:{skill_name}:{tenant_id}:{request_key}"
    if not await redis_client.setnx(dedup_key, "1", ex=60):
        existing = await task_repo.find_by_request_key(skill_name, tenant_id, request_key)
        return existing.task_id
    await task_repo.create_task(
        task_id=task_id,
        request_key=request_key,
        skill_name=skill_name,
        tenant_id=tenant_id,
        user_id=user_id,
        session_id=session_id,
        payload_json=json.dumps(payload, ensure_ascii=False),
        status="PENDING",
        created_at=datetime.now(timezone.utc),
    )
    topic = route_topic_by_skill(skill_name)
    await kafka_producer.send(
        topic=topic,
        value={
            "task_id": task_id,
            "skill_name": skill_name,
            "tenant_id": tenant_id,
            "user_id": user_id,
            "session_id": session_id,
        },
    )
    return task_id

Skill Loader

class SkillLoader:
    def __init__(self, skill_dir: str):
        self.skill_dir = Path(skill_dir)
        self._cache: dict[str, tuple[str, dict]] = {}
        self._lock = asyncio.Lock()

    async def load(self, skill_name: str) -> dict:
        async with self._lock:
            path = self.skill_dir / f"{skill_name}.yaml"
            raw = path.read_text(encoding="utf-8")
            digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
            cached = self._cache.get(skill_name)
            if cached and cached[0] == digest:
                return cached[1]
            spec = yaml.safe_load(raw)
            validate_skill_spec(spec)
            self._cache[skill_name] = (digest, spec)
            return spec

Uses a content hash + in‑memory cache to avoid re‑parsing on every request.

Skill Executor

class SkillExecutor:
    def __init__(self, tool_registry, session_store, audit_repo, llm_gateway):
        self.tool_registry = tool_registry
        self.session_store = session_store
        self.audit_repo = audit_repo
        self.llm_gateway = llm_gateway
        self.llm_semaphore = asyncio.Semaphore(32)

    @asynccontextmanager
    async def _task_guard(self, task_id: str):
        await self.audit_repo.mark_running(task_id)
        try:
            yield
            await self.audit_repo.mark_finished(task_id)
        except Exception as exc:
            await self.audit_repo.mark_failed(task_id, str(exc))
            raise

    async def execute(self, task: dict, skill_spec: dict) -> dict:
        async with self._task_guard(task["task_id"]):
            context = await self.session_store.load(task["session_id"])
            result = await self._run_flow(task, context, skill_spec)
            await self.session_store.save(task["session_id"], result.get("session_summary"))
            return result

    async def _run_flow(self, task: dict, context: dict, skill_spec: dict) -> dict:
        state = {"input": task["payload"], "context": {"request_id": task["request_key"]}, "steps": {}}
        for step in skill_spec["flow"]["steps"]:
            action = step["action"]
            if action == "tool":
                tool_name = step["tool"]
                tool = self.tool_registry.get(tool_name)
                args = render_args(step["args"], state)
                tool_result = await tool.invoke(args)
                state["steps"][tool_name] = {"result": tool_result}
                await self.audit_repo.record_tool_call(task["task_id"], tool_name, args, tool_result)
            elif action == "condition":
                if eval_condition(step["expression"], state):
                    nested = step.get("when_true", [])
                    nested_skill = {"flow": {"steps": nested}}
                    return await self._run_flow(task, context, nested_skill)
            elif action == "respond":
                return render_response(step["data"], state)
        raise RuntimeError("skill flow ended without respond step")

The executor isolates the declarative flow from platform concerns, guarantees audit, and limits concurrent LLM calls via a semaphore.

HTTP Tool Wrapper

class HttpTool:
    def __init__(self, name: str, endpoint: str, timeout_ms: int, idempotent: bool):
        self.name = name
        self.endpoint = endpoint
        self.timeout = timeout_ms / 1000
        self.idempotent = idempotent

    async def invoke(self, args: dict) -> dict:
        headers = {"Content-Type": "application/json"}
        if self.idempotent and "requestId" in args:
            headers["X-Idempotency-Key"] = args["requestId"]
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.post(self.endpoint, json=args, headers=headers)
            resp.raise_for_status()
            return resp.json()

All external calls go through this wrapper, ensuring uniform timeout, idempotency headers, and centralized error handling.

Data Model Design (SQL)

Three tables are required for a production‑grade system.

CREATE TABLE agent_task (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  task_id VARCHAR(64) NOT NULL UNIQUE,
  request_key VARCHAR(64) NOT NULL,
  tenant_id VARCHAR(64) NOT NULL,
  user_id VARCHAR(64) NOT NULL,
  session_id VARCHAR(64) NOT NULL,
  skill_name VARCHAR(64) NOT NULL,
  skill_version VARCHAR(32) DEFAULT NULL,
  status VARCHAR(32) NOT NULL,
  payload_json JSON NOT NULL,
  result_json JSON DEFAULT NULL,
  error_message VARCHAR(1024) DEFAULT NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_tenant_status_created (tenant_id, status, created_at),
  INDEX idx_session_created (session_id, created_at)
);

CREATE TABLE agent_tool_audit (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  task_id VARCHAR(64) NOT NULL,
  tool_name VARCHAR(64) NOT NULL,
  request_json JSON NOT NULL,
  response_json JSON DEFAULT NULL,
  success TINYINT NOT NULL,
  duration_ms INT NOT NULL,
  created_at DATETIME NOT NULL,
  INDEX idx_task_tool (task_id, tool_name)
);

CREATE TABLE agent_manual_review (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  task_id VARCHAR(64) NOT NULL,
  reason_code VARCHAR(64) NOT NULL,
  snapshot_json JSON NOT NULL,
  status VARCHAR(32) NOT NULL,
  reviewer VARCHAR(64) DEFAULT NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_task_status (task_id, status)
);

These tables enable task state recovery, fine‑grained tool audit, and manual takeover tracking.

Observability

Metrics

skill_task_total{skill_name,status}

skill_task_duration_ms_bucket{skill_name}

tool_call_total{tool_name,success}

tool_call_duration_ms_bucket{tool_name}

llm_request_total{model,status}

llm_token_input_total{model}

llm_token_output_total{model}

manual_review_total{skill_name,reason}

Structured Logs

Every log entry includes JSON fields: request_id, task_id, session_id, tenant_id, skill_name, skill_version, tool_name, duration_ms, error_code.

Tracing

A complete trace links API gateway → task creation → skill runtime → LLM gateway → each tool call → database writes, enabling rapid root‑cause analysis.

Security and Permissions

Tool whitelist per skill – a skill can only invoke its declared tools.

Write‑operation classification: read_only, write_low_risk, write_high_risk. High‑risk writes (e.g., refunds) require additional checks, optional manual takeover, and stricter RBAC.

Data minimization – only fields required for the current step are injected into the LLM prompt.

Sensitive data (phone, ID, address) is masked before logging or prompt construction; tokens are never logged.

Versioning, Gray‑Release, and Rollback

Each Skill file contains name, version, published_at, published_by, and a change_log. When a task is created, the current skill_version is stored in the task record, guaranteeing that execution uses the exact version validated at creation time.

Gray‑release strategies include tenant‑based rollout, user‑hash bucketing, and explicit request‑header flags. Automatic rollback is triggered when the new version exceeds thresholds for error rate, manual‑review rate, or latency.

Performance Optimization

Cache read‑only query results with short TTL in Redis.

Cache user profiles and activity configs locally.

Trim unnecessary context from prompts; use structured schemas instead of long prompts.

Rate‑limit LLM calls with a token‑bucket semaphore.

Enable connection pools and circuit‑breakers for downstream HTTP tools.

Separate logging, tracing, and audit pipelines to avoid bottlenecks.

Typical bottleneck distribution: ~30% LLM, ~30% downstream tools, ~20% context assembly, ~10% DB/Cache, ~10% observability overhead.

Common Misconceptions

Zero‑code means no platform code. The platform still needs API, task scheduler, audit, and observability components.

More flexible skills are always better. High‑risk scenarios require strict constraints, not unlimited model freedom.

Register every tool globally. Over‑exposure leads to model confusion, token waste, and security risk.

Write‑heavy flows can stay synchronous. Long‑running writes need task‑based execution with idempotency and rollback handling.

Migration Steps (Zero‑Code Adoption Path)

Select a high‑frequency, low‑risk use case (e.g., order query).

Extract its prompt, tool list, input/output schema, and flow into a Skill YAML file.

Deploy the minimal platform: Skill Loader, Task table, Tool audit.

Introduce Kafka and asynchronous workers for long‑running tasks.

Gradually add gray‑release, rollback, manual takeover, full observability, and fine‑grained permission controls.

This incremental approach avoids a heavyweight upfront build while delivering immediate benefits.

Conclusion

AgentScope 2.0.3’s Skills turn AI assistant capabilities into reusable, versioned, and governable work templates. The platform layer provides safety nets—task isolation, observability, security, and gray‑release—so that AI assistants can evolve at the speed of business without sacrificing reliability. In practice, Skills enable teams to ship new business logic without code releases, isolate high‑risk operations, and maintain a production‑grade, observable AI service.