第 19 章：企业级部署与运维实操：把原理落成生产系统

第 18 章教"怎么写代码"，本章教"怎么上线、怎么运维"。前 6~18 章回答"为什么这样设计才可靠"，本章回答"照着这套原理，实际部署一台可信业务智能体要走哪几步"。全程用可操作的分步说明、配置示例与验证方法，配合流程图便于按图索骥。

图 19-1　企业级部署八步落地流程

flowchart TD
    A[环境准备<br>Python/数据库/密钥] --> B[初始化与降级首跑<br>无 LLM 也先跑通]
    B --> C[配置业务规则<br>第一条硬规则并验证]
    C --> D[接入审批链与权限<br>关键操作人机协同]
    D --> E[可观测与告警<br>日志/追踪/事件订阅]
    E --> F[压测与性能验证<br>延迟预算]
    F --> G[上线演练与回滚<br>灰度+回滚预案]
    G --> H[对照第 18.7 验收清单<br>逐项核对后放行]

19.1 环境准备：把"要什么"先列齐

原则：环境变量先行，密钥绝不写进代码。 生产部署前准备好下表，缺一项都可能在半路翻车，见表 19-1：

表 19-1　环境准备清单

项目	说明	检查要点
Python 运行时	框架基于 Python 标准库，推荐 3.10+	`python --version` 确认
业务数据库	存放会话状态、审计日志	单独只读/读写账号，权限最小化
LLM 接入	模型 API Key 或内网网关	走环境变量注入，不入代码库
密钥管理	API Key / Token / 数据库口令	生产用密钥服务，禁止硬编码
部署目录	代码 + `prog/runtime/` 内嵌框架	确认 `runtime/` 69 个模块齐全
时钟与网络	审计依赖可信时间戳	NTP 校准；内网策略放行模型网关

把表 19-1 的清单落成一张运行拓扑，可看清请求路径、四层框架与三类"可降级外部依赖"的挂接位置，如图 19-2 所示：

图 19-2　企业级部署拓扑：接入侧、四层框架与可降级外部依赖

flowchart TB
    subgraph ACCESS["接入侧"]
        direction LR
        BIZ["业务系统<br/>ERP / OA / 报表"]
        USER["业务人员<br/>Web / IM / 移动端"]
    end

    subgraph APP["智能体服务（Python 进程）"]
        direction TB
        API["HTTP API 层<br/>鉴权 · 限流 · 契约校验"]
        subgraph RUNTIME["prog.runtime 四层"]
            direction LR
            SEM["语义层<br/>意图 · 槽位 · 记忆"]
            ORC["编排层<br/>生命周期 · 流水线 · 流程"]
            GOV["治理层<br/>规则 · 权限 · 审计"]
            INF["基础设施层<br/>日志 · 事件 · 缓存 · 认证"]
        end
        API --> SEM
    end

    subgraph EXT["外部依赖（缺失即降级）"]
        direction LR
        LLM["LLM 网关<br/>缺失 → 规则降级回复"]
        DB["业务数据库<br/>缺失 → 内存链 / 默认配置"]
        REDIS["Redis 事件总线<br/>缺失 → 进程内总线"]
    end

    subgraph OBS["可观测"]
        direction LR
        LOG["结构化日志"]
        TRACE["链路追踪"]
        ALERT["告警订阅"]
    end

    BIZ --> API
    USER --> API
    INF --> LLM
    INF --> DB
    INF --> REDIS
    INF --> LOG
    ORC --> TRACE
    GOV --> ALERT

读图要点： 三类外部依赖（LLM / DB / REDIS）都只与基础设施层相连，且每个都标注了"缺失即降级"的替代实现——这是 16.4 节"每一级能力都有最小可信实现"的部署形态，也是 19.2 节"降级模式首跑"能成立的结构前提。

可观测（日志 / 追踪 / 告警）不是旁路插件，而是由 INF / ORC / GOV 三层在关键路径上主动写入——因此 19.5 节的告警能力在降级模式下依然可用。

19.2 初始化与降级模式首跑

第 1 步：拉起最小服务（无 LLM、无数据库也能启动）

# 进入项目目录，先跑框架自检
python tests/test_framework.py
# 启动最小服务（降级模式，不注入 LLM/数据库）
python -m prog.runtime.server --config config/minimal.yaml

第 2 步：验证"降级模式"行为（这是第 8.5 节"最小可信实现"的工程落地）：

发一条业务咨询 → 走咨询通道（knowledge_assistant 兜底），返回结构化响应契约；
发一条越权操作 → 被规则校验拦截，返回 blocked，而不是"假装执行"；
全程无外部依赖仍可工作——说明正确性验证不依赖外部能力。

第 3 步：注入真实依赖，逐步升级

# 注入 LLM 与数据库，关闭降级开关
python -m prog.runtime.server --config config/prod.yaml

检验：先降级跑通、再注入依赖，顺序本身就是在验证"哪一层坏了也能降级"。若降级模式都跑不通，说明是代码问题，不是配置问题。

19.3 配置业务规则：第一条硬规则

把"金额必须为正、订单号格式必须合法"这类约束固化为硬规则（is_hard=True，bypass=False），模型不可绕过：

from prog.runtime.rule import BaseRule

class AmountPositive(BaseRule):
    def __init__(self):
        super().__init__(
            rule_name="金额必须为正",
            is_hard=True,            # 硬规则：不可绕过
            spec_ref="S-AMT-001",
        )
    def check(self, payload, context):
        amount = payload.get("amount", 0)
        return amount > 0, f"金额 {amount} 必须大于 0"

验证步骤（每一步都要能解释为什么），见表 19-2：

表 19-2　第一条硬规则的验证步骤

验证	操作	预期结果
1 注册	`RuleRegistry.register(AmountPositive())`	注册成功，无报错
2 正向	提交 `amount=100` 的单据	校验通过
3 反向	提交 `amount=-5` 的单据	校验拦截，返回 `blocked` 与原因
4 绕过测试	尝试构造不带该规则的通道	硬规则在所有写操作前执行，无法绕过

要点：规则验证是"生产前最值得花时间的 10 分钟"。每条规则都值得做一次正向 + 一次反向验证，并留档（spec_ref）供审计追溯。

19.4 接入审批链与权限：关键操作人机协同

第 1 步：定义审批流程（定义与实例分离，改流程不影响运行中的实例）：

from prog.runtime.approval import ApprovalChain, Gate
chain = ApprovalChain([
    Gate("发起校验", check=lambda ctx: ctx.user.role in {"manager", "finance"}),
    Gate("金额阈值", check=lambda ctx: ctx.payload["amount"] <= 100_000),
    Gate("复核签字", human=True),   # 人工节点，必须人确认
])

第 2 步：接权限体系（角色 + 属性双重判定）：

from prog.runtime.security import PermissionSystem
perm = PermissionSystem()
perm.grant("query_order", roles=["sales", "manager", "finance"])
perm.grant("create_order", roles=["manager"], attrs={"dept": "sales"})

第 3 步：上线前模拟一次完整流程：发起 → 阈值校验 → 人工确认 → 留痕 → 查审计日志确认"谁批的、依据哪条规则"。审批链没跑通，不接入生产。

19.5 可观测与告警：让系统"自己说话"

统一 trace：全链路单一 trace_id，从请求进入贯穿到审计落库；
结构化日志：按第 16.7.1 节 schema 输出（时间戳、trace_id、agent、事件、结果）；
事件订阅：注册关键事件（审批通过、规则拦截、热更新生效）的订阅者，推送告警：

规则拦截率突增 → 可能是规则误伤，需人工复核；
审批超时未决 → 推送经办人；
模型调用异常 → 触发降级预案。

from prog.runtime.events import bus
bus.subscribe("rule.blocked", lambda e: alert(e))   # 规则拦截告警
bus.subscribe("approval.pending_timeout", lambda e: alert(e))

告警阈值建议（起步值，按运行一周后校准）：规则拦截率周环比波动超过 ±30% 告警；审批平均处理时长超过目标 2 倍告警；模型调用失败率 > 1% 告警。

19.6 压测与性能验证：先证明"快且对"

第 1 步：设延迟预算（从 SLO 倒推，见 17.4.2）。例如"查询接口 P95 < 3 秒"，分解为：规则校验 20ms + 意图识别 200ms + 模型调用 2.5s + 尾部缓冲。

第 2 步：并发正确性验证（正确性优先于性能）：

并发提交同一会话的多轮查询 → 状态互不串扰（第 17.2 章）；
并发写操作 → 互斥生效，无重复入账。

第 3 步：压测与回归：

# 用 ab 或 wrk 打流量，观察 P95 与错误率
ab -n 5000 -c 50 http://localhost:8080/api/query
# 压测后重跑降级模式全量测试，确认性能优化未破坏正确性
python tests/test_framework.py

结论范式：只有"并发下结果依然正确"的压测才有意义——这正是第 17 章标题"正确性是性能的前提"的落地。

19.7 上线演练与回滚：把"出问题怎么办"提前想好

灰度三步：

先接入 1 个低风险场景（如周报查询）→ 观察一周准确率与告警；
再接入 1 个写操作场景（带审批链）→ 观察审批流与审计完整性；
全部平稳后再逐步放量。

回滚预案（演练中至少跑通一次）：

配置回滚：热更新（第 14.4 章）导致规则异常 → 回滚到上一版配置并复核审计；
版本回滚：代码异常 → 切回上一个可运行版本，数据用备份恢复；
降级回滚：模型不可用 → 切降级模式，只读通道保持服务（第 16.4 章）。

上线门禁：对照第 18.7 章验收清单逐项核对（正确性 / 契约 / 路由 / 状态 / 规则 / 安全 / 流程 / 训练 / 记忆 / 可观测 / 性能 / 测试），全绿才放行。

企业质量体系（ISO 9001:2015）与网络安全等级保护（GB/T 22239-2019）对"可追溯、可审计"的要求[21][22]，以及《生成式人工智能服务管理暂行办法》对内容安全与留痕的要求[23]，正对应本节清单中的"安全 / 流程 / 可观测"三项。

19.8 实现剖析：灰度开关、单例并发与"模块关闭即降级"

19.2 节讲"降级模式首跑"，19.7 节讲"灰度三步"。本节从 runtime/module_toggles.py（156 行）摘取三段，看"模块可独立启停、缺模块不影响其余功能"这条运维要求在代码里如何落地。

19.8.1 默认全开 + 未知即放行：不因缺配置阻断业务

# 默认模块开关（所有模块默认启用）
DEFAULT_TOGGLES = {
    "sales_agent": True,
    "production_agent": True,
    "warehouse_agent": True,
    "technical_agent": True,
    "finance_agent": True,
    "knowledge_assistant": True,
    "qc_agent": True,
    "hr_agent": True,
}

def is_enabled(self, module_name: str) -> bool:
    """检查模块是否启用；未知模块默认返回 True。"""
    with self._toggle_lock:                      # ★ 读也要加锁（并发可见性）
        return self._toggles.get(module_name, True)

def _load_from_db(self) -> None:
    """从 system_configs 表加载模块开关配置（可选依赖）。"""
    try:
        from prog.runtime.database import get_database     # ★ 可选依赖：延迟导入
        db = get_database()
        row = db.query_one("system_configs", {"config_key": "module_toggles"})
        if not row:
            return
        raw = row.get("config_value")
        toggles = json.loads(raw) if isinstance(raw, str) else raw
        if not isinstance(toggles, dict):
            return
        with self._toggle_lock:                  # ★ 仅更新已知模块且值为布尔
            for key, value in toggles.items():
                if key in DEFAULT_TOGGLES and isinstance(value, bool):
                    self._toggles[key] = value
    except Exception:
        pass                                     # 无数据库/解析失败 → 静默降级

这段代码印证 19.2 与 19.5 节：运维配置的默认方向同样朝"可用"倾斜。默认全开保证"没配任何东西也能跑"；is_enabled 对未知模块返回 True，使"新增模块而配置表未更新"不会导致功能凭空消失；_load_from_db 用延迟导入 + 宽 except 把"无数据库 / 表不存在 / JSON 非法"统一收敛为静默降级，且只覆盖已知模块的布尔值——这是配置加载的"最小信任面"：只接受自己认识的键，只接受布尔值。与 19.7 节的灰度策略配合，模块开关就是"按模块灰度"的现成开关。

19.8.2 装饰器即降级点：模块关闭时返回统一契约而非抛错

def require_module(module_name: str) -> Callable:
    """模块开关装饰器：模块关闭时被装饰方法自动降级返回统一响应。"""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            manager = ModuleToggleManager.get_instance()
            if not manager.is_enabled(module_name):
                # ★ 模块关闭：返回降级响应，不执行实际 Agent 逻辑
                from prog.runtime.base_agent import AgentResponse
                return AgentResponse(
                    content=f"模块 {module_name} 已关闭，相关功能暂不可用。",
                    action="module_disabled", agent_name=module_name)
            return func(*args, **kwargs)
        return wrapper
    return decorator

这段代码印证 19.2 与 8.2 节：降级要"可被上层识别"，而不是"抛异常让调用方猜"。模块关闭时返回的是与正常路径同一契约（AgentResponse）的响应，只是 action 标记为 module_disabled——上层按 action 分支即可给出正确提示，无需为"模块关闭"单写异常处理。

这是第 8.2 节统一契约在运维场景的收益：降级路径与正常路径共享数据形状，运维开关才不会在业务代码里引发连锁改动。

19.8.3 单例的并发安全：double-checked locking

class ModuleToggleManager:
    _instance: Optional["ModuleToggleManager"] = None
    _singleton_lock = threading.Lock()          # 保护 _instance 创建

    @classmethod
    def get_instance(cls) -> "ModuleToggleManager":
        if cls._instance is None:               # ① 无锁快路径
            with cls._singleton_lock:
                if cls._instance is None:       # ② 锁内二次检查（double-check）
                    cls._instance = cls()
        return cls._instance

    def __init__(self):
        self._toggle_lock = threading.Lock()    # 保护 _toggles 的并发读写
        self._toggles: Dict[str, bool] = dict(DEFAULT_TOGGLES)
        self._load_from_db()

这段代码印证 17.2 节：运维开关是全局单例，必须在多线程下保持一致性。get_instance 用 double-checked locking 兼顾"快路径不加锁"与"创建只发生一次"；__init__ 另为开关字典单独持锁，使运行时 enable / disable 与读取之间不出现撕裂读。

第 17 章"线程本地与加锁"的纪律在运维面同样适用——开关的读写在生产环境天然并发，单例不是"图省事"，而是必须被正确加锁的共享状态。

19.9 本章小结

表 19-3　部署阶段与关键检验

阶段	核心动作	关键检验	对应原理
准备	列齐环境变量与密钥	无硬编码密钥	第 12 章
初始化	降级模式首跑	无 LLM 也正确	第 8 章
规则	第一条硬规则	反向用例被拦截	第 11 章
审批权限	审批链 + 权限	人工节点生效留痕	第 13 / 12 章
可观测	日志 / 追踪 / 告警	单一 trace_id	第 16 章
压测	延迟预算 + 并发	并发下仍正确	第 17 章
上线	灰度 + 回滚 + 验收	验收清单全绿	第 18 章

一句话记住本章：第二部的原理（生成 × 裁决、规则、审批、审计、可观测、并发）不是纸面设计，而是每一步都有可验证操作的部署流程——按本章顺序做完一遍，就等于把 6~18 章的原理全部落到了生产环境。

← 第 18 章：工程实践：从零构建业务智能体附录 A：模块 API 速查 →

规则驱动的业务智能体