AI Agent 生产落地架构：从个人 CLI 到团队平台

flowchart LR
  A["团队入口: Issue / PR / Chat"] --> B["Agent 工作台"]
  B --> C["编排层: Plan / Execute / Verify"]
  C --> D["工具层: Shell / Git / DB / Browser"]
  B --> E["上下文层: 代码图谱 / RAG / 记忆"]
  C --> F["安全层: RBAC / 沙箱 / 审计"]
  D --> G["可观测层: 日志 / 轨迹 / 成本"]
  F --> G
  E --> C

{
  "task_id": "agent-task-2026-0012",
  "source": "github_issue",
  "project": "billing-service",
  "actor": "dev-zhang",
  "intent": "修复发票金额四舍五入错误",
  "priority": "P1",
  "risk_level": "medium",
  "target_branch": "main",
  "acceptance": [
    "新增单元测试覆盖金额边界值",
    "npm run test:billing 必须通过",
    "不能修改 invoice schema"
  ]
}

context_pack:
  project_profile:
    framework: "FastAPI + PostgreSQL"
    language: "Python 3.11"
    test_command: "pytest tests/billing -q"
  task_scope:
    changed_area:
      - "app/billing/"
      - "tests/billing/"
    forbidden_area:
      - "app/auth/"
      - "migrations/"
  files:
    required:
      - "app/billing/invoice.py"
      - "app/billing/money.py"
      - "tests/billing/test_invoice.py"
    optional:
      - "docs/billing-rules.md"
  rules:
    - "金额统一使用 Decimal，不允许 float"
    - "数据库 schema 变更必须人工审批"
  recent_failures:
    - "test_rounding_half_up failed on 19.995"

from dataclasses import dataclass
from enum import Enum


class StepType(str, Enum):
    READ = "read"
    PATCH = "patch"
    COMMAND = "command"
    VERIFY = "verify"
    REPORT = "report"


@dataclass
class AgentStep:
    type: StepType
    title: str
    risk: str
    requires_approval: bool = False


def should_require_approval(step: AgentStep) -> bool:
    risky_keywords = ["deploy", "migration", "delete", "secret", "production"]
    if step.requires_approval:
        return True
    return any(word in step.title.lower() for word in risky_keywords)


plan = [
    AgentStep(StepType.READ, "读取 billing 相关文件", "low"),
    AgentStep(StepType.PATCH, "修复 Decimal 四舍五入逻辑", "medium"),
    AgentStep(StepType.VERIFY, "运行 pytest tests/billing -q", "low"),
    AgentStep(StepType.REPORT, "输出 diff 摘要和测试结果", "low"),
]

for step in plan:
    if should_require_approval(step):
        print(f"WAIT_APPROVAL: {step.title}")
    else:
        print(f"RUN: {step.title}")

security_policy:
  default: deny
  filesystem:
    allow_read:
      - "app/**"
      - "tests/**"
      - "docs/**"
    allow_write:
      - "app/billing/**"
      - "tests/billing/**"
    deny_write:
      - ".env"
      - "migrations/**"
      - "infra/production/**"
  commands:
    allow:
      - "pytest tests/billing -q"
      - "ruff check app/billing tests/billing"
    require_approval:
      - "git push"
      - "docker compose up"
      - "alembic upgrade"
    deny:
      - "rm -rf"
      - "curl * | sh"
      - "chmod 777"
  network:
    allow_hosts:
      - "api.github.com"
    deny_private_network: true

CREATE TABLE agent_events (
  id BIGSERIAL PRIMARY KEY,
  task_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  actor TEXT NOT NULL,
  model TEXT,
  tool_name TEXT,
  input_tokens INTEGER DEFAULT 0,
  output_tokens INTEGER DEFAULT 0,
  cost_usd NUMERIC(12, 6) DEFAULT 0,
  risk_level TEXT DEFAULT 'low',
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_agent_events_task_id ON agent_events(task_id);
CREATE INDEX idx_agent_events_created_at ON agent_events(created_at);