持久化 AI 同事这个概念近几年逐渐从实验室走向了大型企业的生产环境。它指的不是一次性问答机器人而是能够记住历史任务、保留上下文、在流程中被反复调用的 AI Agent。航运物流行业对这类 AI 的需求非常具体因为业务流程长、节点多、单证复杂。以 Maersk马士基这种体量的全球航运物流企业为例每天处理的订舱、提单、报关、锁柜和异常件数据量极大任何一个环节理解错误都可能直接导致集装箱滞留、费用纠纷甚至合规风险。因此这类企业落地“AI 同事”时同时关注两件事纠错能力和模型主权。纠错能力决定模型输出能不能被业务系统信任模型主权决定企业能不能把内部数据、微调权、推理环境和版本演进完整掌握在自己手里。本文会从概念、架构、代码实现、验证、排错到生产落地完整拆解如何构建一个可记录状态、可纠错、可自托管模型的持久化 AI 同事。1. 先理解持久化 AI 同事解决问题的核心机制1.1 从无状态问答到有状态协作者普通的大模型应用是“无状态”的用户发一段问题模型返回一段答案对话结束后系统没有保留任何业务上下文。对于文档问答、内容生成这类场景无状态设计完全够用。但航运物流、工厂排产、财务审核这类场景不一样一个任务通常要跨多个系统、多个时间点才能完成。例如处理一票海运订舱客户先发来订舱委托书。AI 需要读取发运人、收货人、起运港、目的港、箱型、箱量、货名、HS 编码等信息。中途客户补充了“改港申请”仓库端又反馈了“集装箱超重”船司客服再更新了“ETD 延迟一天”。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 这类航运物流企业的典型场景大概是这样的业务场景任务模块职责记忆模块保存内容纠错模块校验项订舱委托解析识别委托书并提取结构化字段客户历史订舱偏好、历史改单记录港口代码、ETD/ETA 日期格式、货名语种提单确认件生成汇总订舱、装箱、报关信息生成草稿各航次运价、客户确认过的模板提单号格式、收发货人地址、币种、总箱量一致性异常件处理判断异常类型并给出处理建议异常历史、仓库反馈记录、船司公告异常类型枚举、赔付金额上限、处理时效这张表说明一个事实无状态 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_asciiFalse), 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_asciiFalse) 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_asciiFalse))这里记录 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_urlself.base_url, api_keyself.api_key) def chat_json(self, messages: list[dict], temperature: float | None None) - str: resp self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature 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, resultparsed, error_detail; .join(errors), ) return {status: waiting_human_review, errors: errors, parsed: parsed} self.memory.update_task_status(task_id, completed, resultparsed) 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 模型调用参数参数建议值说明调大的影响调小的影响temperature0.1控制输出随机性结果更多样但提取类任务容易出错输出更稳定可能过度重复max_tokens1024单次生成上限能处理更长输出占用更多显存/流量长输出会被截断top_p0.9核采样阈值增加多样性偏向高概率词timeout60s模型响应超时容忍慢推理响应延迟变高快速失败但容易误报4.2 状态和记忆参数参数建议值说明history_limit20加载历史消息条数超出部分可以摘要压缩context_window8192 tokens模型上下文窗口超长历史需要截断或检索feedback_ratio30%人工复核通过后进入微调样本的比例避免重复劳动task_timeout3600s任务最大等待时长超时自动标记失败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参数含义参数说明--tensor-parallel-size使用几张 GPU 做张量并行2 表示两张卡--max-model-len最大上下文长度越长占用显存越多--gpu-memory-utilization允许占用的 GPU 显存比例保留一部分给服务进程--served-model-name对外暴露的模型名客户端用这个名称调用学习环境建议从 7B 或 8B 模型开始单张 24G 显存可以跑通生产环境再根据吞吐量扩展到 14B 或 70B 并结合量化方案优化。5. 运行验证正确输入、错误输入、断点恢复5.1 启动最小系统首先启动自托管模型。如果没有本地 GPU也可以临时把 LLM_BASE_URL 指向测试用的 OpenAI 兼容服务先跑通业务逻辑再切回自托管模型验证模型主权链路。export LLM_BASE_URLhttp://localhost:8000/v1 export LLM_MODELinternal-llm export LLM_TEMPERATURE0.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。预期结果statuswaiting_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 消息消失。可能原因SQLite 文件路径不对启动目录不一致导致连接了不同数据库文件。写入后未 commit。数据库被删除或覆盖。排查路径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。排查路径检查 validate_booking 是否真的被调用而不是只做了 JSON 格式化。检查传入 validator 的字段名和 prompt 中定义的字段名是否完全一致。比如 prompt 说 port_of_discharge解析 JSON 时却读 discharge_port规则就永远匹配不上。检查规则配置是否被加载特别是从配置文件读取枚举表时中英文或大小写不一致容易漏掉。解决建议给每条规则增加 rule_id在错误信息里附带规则编号方便定位是哪一条规则拦截。例如[RULE-003] 目的港代码 YYZ 不在港口字典中6.3 自托管模型后效果下降现象从云端模型切换到本地模型后提取字段的准确率明显下降。可能原因本地模型的 instruction-following 能力弱于商用大模型尤其是在输出 JSON 格式时本地小模型可能夹带多余解释。prompt 模板是为云端模型调试的换模型后没有重新适配。采样参数不合适temperature 仍然沿用高随机性设置。解决建议把 model 切换类操作设为独立配置项每次切换后先在小批量真实样本上做回归测试再扩大范围。JSON 输出不稳定时改用结构化输出约束如 OpenAI 兼容的 response_format 或本地支持的 JSON mode。小模型对 prompt 中的“只输出 JSON”更敏感需要把格式说明写得非常具体最好给出一个输出模板。6.4 人工复核与自动纠错重复冲突现象业务员在人工复核界面修改了结果但 AI 的下一次自动纠错又把字段改成了原模型输出。解决建议人工复核的优先级必须高于自动纠错。复核后的结果一旦落库就要把该任务标记为“人工已确认”后续自动流程只能读取不能覆盖。实现方式是在 tasks 表增加 human_confirmed 字段更新前判断该字段。7. 生产环境的工程保障与最佳实践7.1 企业级落地检查清单从 Maersk 式的大型场景抽象落地前至少要把清单逐项核对一遍检查项要求任务追踪每笔 AI 操作必须有全局 task_id支持查询和回放权限隔离AI 只能访问被授权客户、被授权港口、被授权单证类型操作审计模型调用、规则命中、人工复核、反馈回流全部留日志模型版本记录每次响应使用的模型名、版本号、prompt 版本回滚方案模型效果回退时能在配置层面一键切换旧模型数据备份SQLite/PostgreSQL 定期备份反馈表单独导出监控告警校验命中率、人工复核耗时、模型响应延迟都要有指标灰度发布先放开低风险客户或低风险单证类型验证后再扩大范围7.2 模型主权的实践建议模型主权不是一次性工程而是一条持续演进的路线。企业可以从下面三个阶段逐步推进阶段一云端 API 试点但数据脱敏、输出落库、规则拦截全部在线确保数据可追溯。阶段二内部 GPU 集群部署开源模型统一走 OpenAI 兼容接口业务侧无感知切换。阶段三基于内部反馈数据做领域微调或 LoRA 适配沉淀专用模型同时保留云端模型作为备选。每个阶段都要保留“退出通道”数据可以导出、模型可以替换、接口协议可以迁移。这样模型主权才不是一句口号。7.3 扩展方向这篇文章里实现的是一个最小闭环真实项目中还可以继续扩展长期记忆升级为向量检索用 embedding 模型检索“最相关的历史问题”而不是简单拼接最近 N 条消息。纠错规则从硬编码改为规则引擎支持业务人员在界面上配置枚举、正则和联动条件。引入多 Agent 协作一个 Agent 负责单证理解一个 Agent 负责运价校验一个 Agent 负责风险提示最后由主控 Agent 汇总。构建离线评估集每次模型升级前用同一批测试数据跑回归测量字段级准确率而不是只看“整体感觉”。持久化 AI 同事的核心判断只有一条模型可以换状态和纠错机制不能丢。把对话历史、任务状态、人工反馈和企业数据保留在自己的系统里把模型调用做成可以随时切换的标准化接口AI 才能真正从“尝鲜工具”变成“可靠的业务同事”。
