AI Agent 企业应用不是"能聊就行"：集成、权限、审批、审计、成本五道关，附可运行代码

# gateway.py
import httpx
from abc import ABC, abstractmethod

class ToolAdapter(ABC):
    @abstractmethod
    def invoke(self, tenant_id: str, **kwargs) -> dict: ...

class CrmAdapter(ToolAdapter):
    def invoke(self, tenant_id, customer_id, **kw):
        r = httpx.get(f"https://crm.internal/api/customer/{customer_id}",
                      headers={"X-Tenant": tenant_id}, timeout=8)
        r.raise_for_status()
        return {"name": r.json()["name"], "level": r.json()["level"]}

class TicketAdapter(ToolAdapter):
    def invoke(self, tenant_id, title, content, **kw):
        # 老系统只有DB视图，走内部服务封装
        r = httpx.post("http://ticket-svc/internal/create",
                       json={"tenant": tenant_id, "title": title, "content": content},
                       timeout=8)
        return {"ticket_id": r.json()["id"], "status": "created"}

REGISTRY = {"get_customer": CrmAdapter(), "create_ticket": TicketAdapter()}

# authz.py
from functools import wraps

ROLE_PERMS = {
    "support": {"get_customer", "create_ticket"},
    "finance": {"get_customer", "refund"},
    "admin":   {"*"},
}

def authorize(tool_name):
    def deco(fn):
        @wraps(fn)
        def wrapper(ctx, **kwargs):
            role = ctx["role"]; tenant = ctx["tenant_id"]
            perms = ROLE_PERMS.get(role, set())
            if "*" not in perms and tool_name not in perms:
                raise PermissionError(f"角色{role}无权调用{tool_name}")
            # 数据域：强制注入租户过滤，模型无法绕过
            kwargs["tenant_id"] = tenant
            result = fn(ctx, **kwargs)
            # 返回后做数据域二次校验（防越权读）
            assert result.get("tenant_id", tenant) == tenant, "跨租户数据泄漏"
            return result
        return wrapper
    return deco

# approval.py
import json, uuid

def execute_with_approval(ctx, tool_name, args, db):
    if tool_name in {"refund", "delete_user", "send_bulk_notice"}:
        approval_id = str(uuid.uuid4())
        db.approvals.insert_one({
            "approval_id": approval_id, "tenant": ctx["tenant_id"],
            "tool": tool_name, "args": args, "status": "PENDING",
            "agent_state": ctx["serialized_state"],   # 关键：保存Agent上下文
        })
        return {"status": "PENDING_APPROVAL", "approval_id": approval_id}
    return REGISTRY[tool_name].invoke(**args)          # 低危直接执行

def resume_after_approval(approval_id, approved, db):
    rec = db.approvals.find_one({"approval_id": approval_id})
    if rec["status"] != "PENDING": return
    if not approved:
        db.approvals.update_one({"approval_id": approval_id}, {"$set": {"status": "REJECTED"}})
        return
    REGISTRY[rec["tool"]].invoke(**rec["args"])        # 审批通过才执行
    db.approvals.update_one({"approval_id": approval_id}, {"$set": {"status": "APPROVED"}})
    # Agent从持久化状态恢复，继续后续步骤

# audit.py
import re, hashlib

PII = [r"\d{17}[\dXx]", r"1[3-9]\d{9}", r"[\w.]+@[\w.]+"]

def mask(text: str) -> str:
    for p in PII:
        text = re.sub(p, lambda m: hashlib.sha256(m.group().encode()).hexdigest()[:8], text)
    return text

def audit_log(span: dict):
    # 每步落：租户、角色、工具、参数摘要(脱敏)、结果状态、耗时、token
    span["args"] = mask(json.dumps(span.get("args", {}), ensure_ascii=False))
    span["result_summary"] = mask(str(span.get("result"))[:200])
    db.audit.insert_one(span)

# cost.py
BUDGET = {"tenant_daily": 50.0}   # 美元
PRICE = {"gpt-4o": 0.005, "gpt-4o-mini": 0.0005}

def route_model(tenant_id, task_complexity):
    spent = db.usage.aggregate_today(tenant_id)
    if spent > BUDGET["tenant_daily"] * 0.8:      # 超80%降级
        return "gpt-4o-mini"
    return "gpt-4o" if task_complexity == "high" else "gpt-4o-mini"

def check_circuit(tenant_id):
    fails = db.metrics.recent_fail_rate(tenant_id, window="5m")
    if fails > 0.3:
        raise RuntimeError("熔断：近期失败率过高，转人工")

# eval_enterprise.py
CASES = [{"goal":"查客户C001等级","expect_tool":"get_customer","expect_kw":"VIP"}]

def evaluate(agent_fn, tenant):
    hit = tool_ok = 0
    for c in CASES:
        ans, trace = agent_fn(c["goal"], tenant=tenant)
        hit += c["expect_kw"] in ans
        tool_ok += any(t.get("tool")==c["expect_tool"] for t in trace)
    return {"answer_hit": hit/len(CASES), "tool_acc": tool_ok/len(CASES)}