从无状态到有状态：构建可纠错、自托管模型的持久化AI同事

持久化 AI 同事这个概念，近几年逐渐从实验室走向了大型企业的生产环境。它指的不是一次性问答机器人，而是能够记住历史任务、保留上下文、在流程中被反复调用的 AI Agent。航运物流行业对这类 AI 的需求非常具体，因为业务流程长、节点多、单证复杂。以 Maersk（马士基）这种体量的全球航运物流企业为例，每天处理的订舱、提单、报关、锁柜和异常件数据量极大，任何一个环节理解错误都可能直接导致集装箱滞留、费用纠纷甚至合规风险。因此，这类企业落地“AI 同事”时，同时关注两件事：纠错能力和模型主权。纠错能力决定模型输出能不能被业务系统信任；模型主权决定企业能不能把内部数据、微调权、推理环境和版本演进完整掌握在自己手里。本文会从概念、架构、代码实现、验证、排错到生产落地，完整拆解如何构建一个可记录状态、可纠错、可自托管模型的持久化 AI 同事。

1. 先理解持久化 AI 同事解决问题的核心机制

1.1 从无状态问答到有状态协作者

普通的大模型应用是“无状态”的：用户发一段问题，模型返回一段答案，对话结束后，系统没有保留任何业务上下文。对于文档问答、内容生成这类场景，无状态设计完全够用。但航运物流、工厂排产、财务审核这类场景不一样，一个任务通常要跨多个系统、多个时间点才能完成。

例如处理一票海运订舱：

1. 客户先发来订舱委托书。
2. AI 需要读取发运人、收货人、起运港、目的港、箱型、箱量、货名、HS 编码等信息。
3. 中途客户补充了“改港申请”，仓库端又反馈了“集装箱超重”，船司客服再更新了“ETD 延迟一天”。
4. AI 如果需要同时记住这些变更，并在最终生成提单确认件时把历史信息全部纳入判断，就必须是有状态的。

持久化 AI 同事，本质上是在 Agent 外面加了一层“业务记忆系统”。它把每一次对话、每一个模型调用、每一条用户反馈、每一笔业务变更都写入持久化存储，下次处理任务时通过检索或拼接重新组织上下文。这样 AI 就从“回答问题的工具”变成了“参与流程的同事”。

1.2 纠错能力为什么是 AI 同事的刚需

大模型生成内容天然存在“幻觉”问题。幻觉在客服聊天里可能只是回答不准确，在航运单证场景里就是实打实的业务事故。比如把目的港代码 AMS 误认为阿姆斯特丹没问题，但如果把“AMS”识别成“安曼”就会导致整套运输计划出错。

纠错不是简单地在模型后面加一个“请检查”指令，而是要在系统层面建立多条防线：

• 规则校验层：港口代码、币种、日期、集装箱尺寸这些字段，可以在模型输出后立即用枚举表、正则或业务字典校验。
• 人工复核层：高敏感操作必须进入人工审批队列，AI 先给出草稿，业务员确认后才落地。
• 反馈回流层：人工修改结果要写回样本库，用于后续微调或 Few-shot 示例选择，让模型逐步学会该场景的正确格式。
• 审计追踪层：每次纠错都要留下记录，方便复盘错误率、模型版本和业务规则变化带来的影响。

这套机制在 Maersk 这类大型航运物流企业的场景中非常典型：错误单证引发的改单成本、港口滞留成本、客户赔偿成本都很高，企业宁可让 AI 先跑、人工再确认，也不愿意让模型直接写死业务数据。落到技术实现上，就是设计一个“模型输出管道 + 校验器 + 人工审核队列 + 反馈存储”的闭环。

1.3 模型主权：企业使用 AI 的技术底线

模型主权（Model Sovereignty）指的是企业对模型资产拥有完整控制权，包括训练数据的所有权、微调权、推理环境的自主掌控权、模型版本的退役权。企业如果只是通过公有云 API 调用大模型，第一是内部数据会被送入第三方服务，第二是模型行为不可审计，第三是无法针对业务场景做深度适配。

过去两年，很多物流、制造、金融企业开始放弃“全 API 调用”，转而采用混合架构：

• 低敏感场景（内部知识问答、文案生成）可以继续使用外部 API。
• 高敏感场景（单证理解、合同审核、运价计算）部署自托管模型，例如在内部 GPU 集群上用 vLLM 部署开源模型，或者运行基于内部数据微调的领域模型。
• 模型接口层统一采用 OpenAI 兼容协议，方便在本地模型和云端模型之间切换。

模型主权不是“必须从零训练一个大模型”，而是企业需要具备“随时可以替换模型、迁移数据、复现结果”的能力。这样即便外部模型提供方调整价格、下架版本或者改变服务协议，企业业务也不会被绑架。

2. 持久化 AI 同事的总体架构设计

2.1 四个核心模块：状态、记忆、任务、纠错

一个可落地的持久化 AI 同事，按职责可以拆成四个模块。

第一个是任务管理模块。它接收外部业务系统发来的任务请求，为每个任务分配全局唯一的 task_id，维护任务状态机（待处理、处理中、等待人工确认、已完成、已失败）。任务管理保证了 AI 的工作可以被追踪和恢复。

第二个是记忆模块。记忆分为短期记忆和长期记忆。短期记忆指当前任务上下文，比如客户发来的三份邮件、两份附件、一次改港申请；长期记忆指跨任务的沉淀，比如这个客户历史上是否经常出现同一类问题、本月是否有运价变更记录。记忆模块可以用 SQLite、PostgreSQL 或向量数据库实现。

第三个是模型调用层。这一层必须做接口抽象，所有模型调用统一走同一个协议，底层可以对接自托管模型，也可以对接云端模型。这样模型主权才能落实在代码层面，而不是停留在口头。

第四个是纠错模块。纠错模块接收模型输出，执行字段校验、规则检查、危险操作判定，如果通过则放行到业务系统，如果不通过则转入人工复核队列。纠错模块还会把人工修改结果记录到反馈表，形成闭环。

2.2 控制流、数据流与模型调用的边界

这四个模块在运行时要清晰地分开控制流和数据流。控制流描述的是“任务怎么被一步步处理”，数据流描述的是“数据在哪个环节被写入、读取、转换和校验”。

推荐的控制流顺序：

接收任务 -> 从记忆模块加载上下文 -> 组装 Prompt -> 调用模型 -> 得到原始输出 -> 规则校验 -> (通过) 写入业务结果 -> (不通过) 进入人工复核队列 -> 人工确认/修改 -> 写回记忆模块 -> 更新任务状态

这里最容易被忽略的边界问题有两个。

第一，模型调用不能直接访问业务库。模型输入和输出都必须先经过“数据转换层”，把业务字段转成模型能理解的文本格式，模型输出再被解析成结构化的 JSON。不要让模型直接背负查询数据库的权限，否则一旦 Agent 权限被滥用，风险面会扩大。

第二，纠错模块不能吞掉错误信息。规则校验失败时，不仅要拦截错误输出，还要把错误字段、期望值、实际值、模型原始输出、上下文片段全部记录下来。没有这些信息，后续优化 prompt、微调数据集和排错都无从下手。

2.3 Maersk 场景落地的架构映射

把上面的架构映射到 Maersk 这类航运物流企业的典型场景，大概是这样的：

这张表说明一个事实：无状态 AI 负责“看懂一句话”，有状态的 AI 负责“推动一个流程”，而纠错模块决定“这个流程是否可信”。企业如果只做了前两层，AI 只能算试运行；加上纠错和状态持久化，才算真正把 AI 变成了团队的一分子。

3. 用 Python 实现最小持久化 AI 同事

3.1 环境准备与依赖

为了快速跑通最小闭环，这里选用 Python 3.10、FastAPI、SQLite 和一个 OpenAI 兼容的模型客户端。模型端既可以是本地部署的 vLLM 服务，也可以是兼容 OpenAI 协议的线上服务，实现时通过 base_url 切换。

python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install fastapi uvicorn openai pydantic

说明：这里没有引入大型 Agent 框架，目的只有一个——把“持久化 + 纠错 + 模型主权”这三个关键机制用最少的代码讲清楚。实际项目中如果团队已经熟悉 LangGraph 或自研 Agent 框架，可以在保留核心模块边界的前提下迁移。

3.2 长期记忆模块：会话状态与结构化存储

先用 SQLite 建立两张表：messages 保存每次模型交互记录，tasks 保存业务任务状态。这样 AI 即使进程重启，也能通过 task_id 恢复上下文。

import json import sqlite3 from datetime import datetime, timezone class AgentMemory: def __init__(self, db_path: str = "agent_to_memory.db"): self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row self.conn.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL ) """) self.conn.execute(""" CREATE TABLE IF NOT EXISTS tasks ( task_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, status TEXT NOT NULL, payload TEXT NOT NULL, result TEXT, error_detail TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """) self.conn.commit() def save_message(self, task_id: str, role: str, content: str): now = datetime.now(timezone.utc).isoformat() self.conn.execute( "INSERT INTO messages(task_id, role, content, created_at) VALUES(?,?,?,?)", (task_id, role, content, now), ) self.conn.commit() def load_messages(self, task_id: str, limit: int = 20): rows = self.conn.execute( "SELECT role, content FROM messages WHERE task_id=? ORDER BY id DESC LIMIT ?", (task_id, limit), ).fetchall() return [{"role": r["role"], "content": r["content"]} for r in reversed(rows)] def create_task(self, task_id: str, agent_id: str, payload: dict): now = datetime.now(timezone.utc).isoformat() self.conn.execute( "INSERT INTO tasks(task_id, agent_id, status, payload, created_at, updated_at) VALUES(?,?,?,?,?,?)", (task_id, agent_id, "pending", json.dumps(payload, ensure_ascii=False), now, now), ) self.conn.commit() def update_task_status(self, task_id: str, status: str, result: dict = None, error_detail: str = None): now = datetime.now(timezone.utc).isoformat() result_str = json.dumps(result, ensure_ascii=False) if result is not None else None self.conn.execute( "UPDATE tasks SET status=?, result=?, error_detail=?, updated_at=? WHERE task_id=?", (status, result_str, error_detail, now, task_id), ) self.conn.commit()

关键点：SQLite 在单机学习场景够用，但生产环境建议换成 PostgreSQL 并使用 JSONB 字段存储 payload 和 result。任务表和消息表分开设计，是因为任务状态和对话上下文的生命周期不同，任务可以完成，消息还需要留作审计。

3.3 纠错回路：规则拦截与人工复核

纠错模块的核心是一个 validator，它接收模型输出的结构化数据，返回校验结果。这里以订舱字段校验为例。

import re from datetime import datetime VALID_PORTS = {"SHA", "NGB", "AMS", "HAM", "SIN", "LKH"} VALID_SIZES = {"20GP", "40GP", "40HQ", "45HQ"} VALID_CUURENCIES = {"CNY", "USD", "EUR", "JPY"} def validate_booking(data: dict) -> list[str]: errors = [] loading_port = data.get("port_of_loading", "") if not loading_port: errors.append("port_of_loading 缺失") elif loading_port.upper() not in VALID_PORTS: errors.append(f"起运港代码 {loading_port} 不在港口字典中") discharge_port = data.get("port_of_discharge", "") if not discharge_port: errors.append("port_of_discharge 缺失") elif discharge_port.upper() not in VALID_PORTS: errors.append(f"目的港代码 {discharge_port} 不在港口字典中") if loading_port and discharge_port and loading_port == discharge_port: errors.append("起运港和目的港不能相同") for field in ["etd", "eta"]: raw = data.get(field, "") if raw: try: datetime.strptime(raw, "%Y-%m-%d") except ValueError: errors.append(f"{field} 日期格式应为 YYYY-MM-DD") container_size = data.get("container_size", "") if container_size and container_size.upper() not in VALID_SIZES: errors.append(f"集装箱尺寸 {container_size} 不在箱型枚举中") quantity = data.get("quantity", 0) if not isinstance(quantity, int) or quantity <= 0 or quantity > 500: errors.append(f"箱量 {quantity} 超出合理范围") currency = data.get("currency", "") if currency and currency.upper() not in VALID_CUURENCIES: errors.append(f"币种 {currency} 不在支持清单中") return errors

在真实项目中，这些规则会从代码里抽出来，放到规则引擎或配置中心，方便业务人员调整。规则校验的位置很讲究：一定要在模型输出被写入业务库之前执行，同时在模型调用之后执行，中间不允许有任何跳过逻辑。

执行流程里，校验通过则直接完成；校验失败则进入人工复核队列。复核完成后的结果需要写回反馈表：

class FeedbackStore: def __init__(self, memory: AgentMemory): self.memory = memory def record(self, task_id: str, model_output: dict, human_result: dict, reason: str): feedback = { "task_id": task_id, "model_output": model_output, "human_result": human_result, "reason": reason, "created_at": datetime.now(timezone.utc).isoformat(), } self.memory.save_message(task_id, "human_feedback", json.dumps(feedback, ensure_ascii=False))

这里记录 model_output 和 human_result 的差异非常关键。后续微调模型时，可以把这些记录转成“错误样本 + 正确输出”的指令数据，也可以把差异明显的样例加入 Few-shot 示例，让模型在下次遇到类似单据时直接输出正确格式。

3.4 模型调用层：为模型主权保留替换入口

模型调用层不直接 import 某个具体模型 SDK，而是统一使用 OpenAI 兼容客户端的 base_url 和 model 参数。这样底层切换自托管模型还是云端模型，只需要改配置，不用改业务代码。

import os from openai import OpenAI class ModelClient: def __init__(self): self.base_url = os.getenv("LLM_BASE_URL", "http://localhost:8000/v1") self.api_key = os.getenv("LLM_API_KEY", "EMPTY") self.model = os.getenv("LLM_MODEL", "qwen/Qwen2.5-14B-Instruct") self.temperature = float(os.getenv("LLM_TEMPERATURE", "0.1")) self.client = OpenAI(base_url=self.base_url, api_key=self.api_key) def chat_json(self, messages: list[dict], temperature: float | None = None) -> str: resp = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature if temperature is not None else self.temperature, ) return resp.choices[0].message.content

把 temperature 默认设为 0.1，是因为单证提取这类任务需要确定性，不需要创造性。越接近 0，输出越稳定；如果任务是多轮解释或方案生成，可以临时调高到 0.3 到 0.7。

3.5 Agent 主流程串联

最后把记忆、模型、纠错串联成一个可运行的 Agent 类：

import json class PersistentAIWorker: def __init__(self, agent_id: str, memory: AgentMemory, model: ModelClient): self.agent_id = agent_id self.memory = memory self.model = model def process_booking_task(self, task_id: str, raw_text: str): self.memory.create_task(task_id, self.agent_id, {"raw_text": raw_text}) history = self.memory.load_messages(task_id) system_prompt = ( "你是一名航运物流订舱助理。请从客户委托文本中提取订舱字段，" "只输出 JSON，不要输出多余解释。字段包括：port_of_loading, " "port_of_discharge, etd, eta, container_size, quantity, cargo_name, " "currency, shipper, consignee。" ) messages = [{"role": "system", "content": system_prompt}] + history messages.append({"role": "user", "content": raw_text}) self.memory.save_message(task_id, "user", raw_text) raw_output = self.model.chat_json(messages) self.memory.save_message(task_id, "assistant", raw_output) parsed = self._safe_parse_json(raw_output) errors = validate_booking(parsed) if errors: self.memory.update_task_status( task_id, "waiting_human_review", result=parsed, error_detail="; ".join(errors), ) return {"status": "waiting_human_review", "errors": errors, "parsed": parsed} self.memory.update_task_status(task_id, "completed", result=parsed) return {"status": "completed", "parsed": parsed} @staticmethod def _safe_parse_json(raw: str) -> dict: try: return json.loads(raw.strip().strip("`").replace("json", "", 1)) except json.JSONDecodeError: return {"_raw": raw}

这样一个最小闭环就完成了：接收文本、存历史、调模型、校验、落库。后面再叠加 FastAPI 接口就能对外提供服务。

4. 关键参数与配置速查

4.1 模型调用参数

4.2 状态和记忆参数

4.3 自托管模型部署参数

如果走模型主权路线，在内部 GPU 集群用 vLLM 部署开源模型，这是常见的启动方式：

python -m vllm.entrypoints.openai.api_server \ --model /data/models/Qwen2.5-14B-Instruct \ --served-model-name internal-llm \ --tensor-parallel-size 2 \ --max-model-len 8192 \ --gpu-memory-utilization 0.85 \ --port 8000

参数含义：

学习环境建议从 7B 或 8B 模型开始，单张 24G 显存可以跑通；生产环境再根据吞吐量扩展到 14B 或 70B 并结合量化方案优化。

5. 运行验证：正确输入、错误输入、断点恢复

5.1 启动最小系统

首先启动自托管模型。如果没有本地 GPU，也可以临时把 LLM_BASE_URL 指向测试用的 OpenAI 兼容服务，先跑通业务逻辑，再切回自托管模型验证模型主权链路。

export LLM_BASE_URL=http://localhost:8000/v1 export LLM_MODEL=internal-llm export LLM_TEMPERATURE=0.1 python -c "from demo import PersistentAIWorker; print('ready')"

然后编写一个测试脚本，构造三种输入：

• 正确输入：字段完整、格式正确，预期校验通过。
• 错误输入：港口代码写错或日期格式错误，预期进入人工复核。
• 中断恢复：任务执行一半后重启进程，再次读取任务时仍能加载历史消息。

5.2 功能验证：正确与错误输入各走各的分支

正确输入示例：

{ "port_of_loading": "SHA", "port_of_discharge": "AMS", "etd": "2025-09-10", "eta": "2025-09-28", "container_size": "40HQ", "quantity": 2, "cargo_name": "textile", "currency": "USD", "shipper": "ABC Trading", "consignee": "XYZ Logistics" }

错误输入示例：

{ "port_of_loading": "SHA", "port_of_discharge": "YYZ", "etd": "2025/09/10", "container_size": "40HQ", "quantity": 0 }

第二个输入的目的港 YYZ 不在港口字典中，日期格式也不规范，quantity 为 0。预期结果：

status=waiting_human_review errors=["目的港代码 YYZ 不在港口字典中", "etd 日期格式应为 YYYY-MM-DD", "箱量 0 超出合理范围"]

5.3 验证持久化能力：任务中断后能否恢复

验证持久化最简单的方式是分两次运行。第一次创建一个 task_id 为 “task-001” 的任务，并写入一条用户消息，然后立即退出进程。第二次重新连接同一个 SQLite 文件，加载该任务的全部历史消息，继续追加新的用户请求。

python check_recovery.py

如果 messages 表中能够查到刚才保存的内容，说明状态持久化生效。这个能力在生产环境对应的是“进程重启、任务不丢、人工正在审核的草稿不消失”。

6. 常见问题与排查路径

6.1 重启后任务状态丢失

现象：进程重启后，已经创建的任务查不到，或者 assistant 消息消失。

可能原因：

1. SQLite 文件路径不对，启动目录不一致导致连接了不同数据库文件。
2. 写入后未 commit。
3. 数据库被删除或覆盖。

排查路径：

sqlite3 agent_to_memory.db "select task_id, status, created_at from tasks;" sqlite3 agent_to_memory.db "select count(*) from messages;"

解决建议：数据库路径使用绝对路径或统一通过配置项注入，避免因工作目录不同产生多份数据库。写入语句统一使用 commit 后返回成功，不要依赖垃圾回收自动提交。

6.2 校验规则不生效

现象：模型输出了错误的港口代码，但系统仍然把结果标记为 completed。

排查路径：

1. 检查 validate_booking 是否真的被调用，而不是只做了 JSON 格式化。
2. 检查传入 validator 的字段名和 prompt 中定义的字段名是否完全一致。比如 prompt 说 port_of_discharge，解析 JSON 时却读 discharge_port，规则就永远匹配不上。
3. 检查规则配置是否被加载，特别是从配置文件读取枚举表时，中英文或大小写不一致容易漏掉。

解决建议：给每条规则增加 rule_id，在错误信息里附带规则编号，方便定位是哪一条规则拦截。例如：

[RULE-003] 目的港代码 YYZ 不在港口字典中

6.3 自托管模型后效果下降

现象：从云端模型切换到本地模型后，提取字段的准确率明显下降。

可能原因：

1. 本地模型的 instruction-following 能力弱于商用大模型，尤其是在输出 JSON 格式时，本地小模型可能夹带多余解释。
2. prompt 模板是为云端模型调试的，换模型后没有重新适配。
3. 采样参数不合适，temperature 仍然沿用高随机性设置。

解决建议：

• 把 model 切换类操作设为独立配置项，每次切换后先在小批量真实样本上做回归测试，再扩大范围。
• JSON 输出不稳定时，改用结构化输出约束（如 OpenAI 兼容的 response_format 或本地支持的 JSON mode）。
• 小模型对 prompt 中的“只输出 JSON”更敏感，需要把格式说明写得非常具体，最好给出一个输出模板。

6.4 人工复核与自动纠错重复冲突

现象：业务员在人工复核界面修改了结果，但 AI 的下一次自动纠错又把字段改成了原模型输出。

解决建议：人工复核的优先级必须高于自动纠错。复核后的结果一旦落库，就要把该任务标记为“人工已确认”，后续自动流程只能读取不能覆盖。实现方式是在 tasks 表增加 human_confirmed 字段，更新前判断该字段。

7. 生产环境的工程保障与最佳实践

7.1 企业级落地检查清单

从 Maersk 式的大型场景抽象，落地前至少要把清单逐项核对一遍：

7.2 模型主权的实践建议

模型主权不是一次性工程，而是一条持续演进的路线。企业可以从下面三个阶段逐步推进：

1. 阶段一：云端 API 试点，但数据脱敏、输出落库、规则拦截全部在线，确保数据可追溯。
2. 阶段二：内部 GPU 集群部署开源模型，统一走 OpenAI 兼容接口，业务侧无感知切换。
3. 阶段三：基于内部反馈数据做领域微调或 LoRA 适配，沉淀专用模型，同时保留云端模型作为备选。

每个阶段都要保留“退出通道”：数据可以导出、模型可以替换、接口协议可以迁移。这样模型主权才不是一句口号。

7.3 扩展方向

这篇文章里实现的是一个最小闭环，真实项目中还可以继续扩展：

• 长期记忆升级为向量检索，用 embedding 模型检索“最相关的历史问题”而不是简单拼接最近 N 条消息。
• 纠错规则从硬编码改为规则引擎，支持业务人员在界面上配置枚举、正则和联动条件。
• 引入多 Agent 协作：一个 Agent 负责单证理解，一个 Agent 负责运价校验，一个 Agent 负责风险提示，最后由主控 Agent 汇总。
• 构建离线评估集，每次模型升级前用同一批测试数据跑回归，测量字段级准确率而不是只看“整体感觉”。

持久化 AI 同事的核心判断只有一条：模型可以换，状态和纠错机制不能丢。把对话历史、任务状态、人工反馈和企业数据保留在自己的系统里，把模型调用做成可以随时切换的标准化接口，AI 才能真正从“尝鲜工具”变成“可靠的业务同事”。