多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

AI Native架构落地实战:从模型接入到状态机编排的工程化设计

AI Native架构落地实战:从模型接入到状态机编排的工程化设计 1. 为什么现在要聊 AI Native 架构过去两年我参与过三个从零起步的 AI 项目也接手过两个传统系统加挂大模型的改造项目。这两类项目的体感差异非常大前者像在平地上盖房子后者像在已经装修好的房子里改承重墙。所以当有人问我AI Native 架构到底该怎么落地时我一般不会先讲技术栈而是先讲一个判断标准——你的系统里AI 是主结构还是外挂件。AI Native 这个词被用得很泛但落到工程上其实很具体系统的核心链路、数据流向、状态管理、失败处理都是围绕模型推理这个不确定过程来设计的。传统架构里一个函数调用要么成功要么失败边界清晰而 AI Native 架构里一次推理可能返回格式不对、可能超时、可能给出一个看起来合理但事实错误的答案。这些不是异常而是常态。架构要做的是把这种不确定性收敛成可观测、可重试、可降级的工程能力。这篇文章适合三类人看一是准备从零搭一个 AI 应用的开发者二是正在做传统系统智能化改造的工程师三是需要判断技术方案是否靠谱的技术负责人。我会按整体设计思路—核心细节—实操落地—问题排查的顺序展开中间会穿插我自己踩过的坑和实际参数选择过程。不堆概念尽量给能直接抄的配置和代码。2. 整体架构设计与思路拆解2.1 先想清楚 AI Native 和AI 加挂的本质区别我见过最常见的误区是把 AI Native 理解成用上大模型 API。这两者差得很远。一个电商系统调用大模型生成商品描述这是 AI 加挂而一个系统把理解用户意图作为入口、把规划任务步骤作为调度核心、把生成结果作为输出层这才是 AI Native。区别体现在三个层面。控制流上传统系统是代码决定下一步做什么AI Native 是模型输出决定下一步做什么代码负责约束和校验。数据流上传统系统数据是结构化、确定性的AI Native 里大量数据是非结构化的文本、向量、多模态内容。失败处理上传统系统靠异常捕获AI Native 靠置信度评估、重试策略和人工兜底。我一般建议团队先画一张图把系统里所有需要判断的环节标出来。如果这些判断点超过一半由模型完成那基本可以按 AI Native 来设计如果只是零星几个那老老实实做加挂改造更划算。这个判断能省掉大量过度设计的成本。2.2 分层设计把不确定性关进笼子AI Native 架构我习惯分成五层从下往上依次是模型接入层、能力编排层、状态与记忆层、业务逻辑层、交互层。这个分法的核心目的是隔离不确定性——越往下越稳定越往上越灵活。模型接入层负责统一不同模型的调用协议。这里的关键是抽象出一个LLMProvider接口把 OpenAI 风格、Claude 风格、本地推理服务的差异屏蔽掉。我实测下来如果一开始不做这层抽象后期换模型或加模型会非常痛苦因为调用参数、返回结构、流式格式都不一样。能力编排层是 AI Native 的心脏。它决定一次请求要调用哪些模型、按什么顺序、要不要并行、失败了怎么重试。这一层我强烈建议用显式的状态机或工作流引擎来实现而不是写成一堆 if-else。原因很简单AI 链路的执行路径是动态的用代码硬编码会迅速失控。状态与记忆层处理上下文。短期记忆是当前会话的消息历史长期记忆是向量库里的知识。这里有个容易忽略的点上下文不是越多越好。我做过对比测试把 20 轮历史全塞进去回答质量反而不如只保留最近 5 轮加摘要因为噪声会干扰模型判断。业务逻辑层是传统代码的主场负责权限、计费、审计这些确定性逻辑。交互层处理流式输出、多轮对话、前端状态同步。这两层相对成熟不多展开。2.3 技术选型为什么我倾向轻框架 显式编排市面上的 AI 框架很多从 LangChain 到各种 Agent 框架。我的经验是原型阶段用框架提效生产阶段要能看懂框架在干什么。很多框架把编排逻辑藏在抽象后面出问题时排查成本极高。所以我的选型倾向是模型接入用轻量 SDK 或直接 HTTP编排用显式的工作流可以是代码里的状态机也可以是 Temporal 这类工作流引擎向量检索用成熟的向量库。这样每一层的边界清晰出问题能快速定位。具体到参数模型接入层我会配置三个关键值timeout设为 30 秒流式场景可放宽到 60 秒max_retries设为 2 次retry_backoff用指数退避从 1 秒起。这三个值不是拍脑袋是实测出来的——超时太短会误杀正常的长推理太长会拖垮整个请求链路重试超过 2 次基本说明是系统性问题再试也是浪费。3. 核心细节解析与实操要点3.1 模型接入层的抽象设计先看接口设计。我一般定义这样一个协议from abc import ABC, abstractmethod from typing import AsyncIterator class LLMProvider(ABC): abstractmethod async def complete(self, messages: list[dict], **kwargs) - str: ... abstractmethod async def stream(self, messages: list[dict], **kwargs) - AsyncIterator[str]: ... abstractmethod def count_tokens(self, text: str) - int: ...count_tokens这个方法经常被忽略但它非常重要。AI Native 系统的成本大头是 token如果不能在调用前估算 token 数就没法做预算控制和上下文裁剪。不同模型的 tokenizer 不一样所以这个方法必须放在 provider 层实现。注意不要用字符数除以 4 来估算 token中文场景下误差能到 50% 以上。要么用对应模型的 tokenizer要么用 tiktoken 这类库但要注意它只对特定模型准确。接入层还要处理一个现实问题不同模型的返回格式差异。有的返回content字段有的返回message.content流式返回的 chunk 结构也不一样。我的做法是在 provider 内部统一转换成标准格式上层完全感知不到差异。这样换模型时只改 provider不动业务代码。3.2 编排层用状态机管住动态路径编排层最容易写乱。我见过一个项目把判断意图—检索知识—生成回答—校验格式全写在一个函数里两百多行加个新分支就要动全身。后来重构成状态机清晰多了。状态机的核心是定义清楚状态和转移条件。一个典型的问答链路状态包括INTENT_PARSING、RETRIEVING、GENERATING、VALIDATING、DONE、FALLBACK。每个状态有明确的输入输出转移条件基于模型输出或校验结果。class WorkflowState(Enum): INTENT_PARSING intent_parsing RETRIEVING retrieving GENERATING generating VALIDATING validating DONE done FALLBACK fallback TRANSITIONS { WorkflowState.INTENT_PARSING: { need_retrieval: WorkflowState.RETRIEVING, direct_answer: WorkflowState.GENERATING, unclear: WorkflowState.FALLBACK, }, WorkflowState.RETRIEVING: { found: WorkflowState.GENERATING, not_found: WorkflowState.FALLBACK, }, WorkflowState.GENERATING: { valid: WorkflowState.DONE, invalid: WorkflowState.FALLBACK, }, }这样设计的好处是每个状态的逻辑独立可测新增分支只需加转移规则。我实测下来这种结构让排查问题的效率提升很明显——出问题时能直接定位到卡在哪个状态。3.3 状态与记忆上下文窗口的取舍艺术上下文管理是 AI Native 架构里最考验经验的部分。模型上下文窗口越来越大但不代表应该全塞进去。我做过一组对比同一个客服场景方案 A 塞全部 20 轮历史方案 B 只保留最近 5 轮加前文摘要方案 B 的答案准确率反而高 12 个百分点。原因是长上下文里混入了大量无关信息模型注意力被稀释。所以我的策略是分层记忆最近 3 到 5 轮保留原文更早的压缩成摘要关键事实抽取成结构化字段单独存。摘要的生成也有讲究。不要每轮都重新摘要那样成本高且容易累积误差。我的做法是每积累 5 轮触发一次摘要把新摘要和旧摘要合并。合并时用这样的提示词请将以下两段对话摘要合并为一段保留所有关键事实、用户偏好和未解决的问题去除重复和寒暄内容。输出不超过 200 字。提示摘要一定要保留未解决的问题这是多轮对话里最容易丢的信息。丢了之后用户会反复重复体验很差。向量检索这块我建议至少配置两个参数top_k和score_threshold。top_k一般设 3 到 5太多会引入噪声score_threshold根据实际数据调我一般从 0.7 起调低于这个分数的结果直接丢弃宁可让模型说不知道也不要给错误答案。3.4 业务逻辑层确定性代码不可替代AI Native 不代表所有逻辑都交给模型。权限校验、计费、审计、数据脱敏这些必须用确定性代码。我见过把权限判断也交给模型的方案结果模型偶尔通融一下直接造成越权。我的原则是凡是能用规则明确表达的就不要用模型。模型只处理规则难以覆盖的模糊判断。比如用户是否有权访问这个文档用 ACL 判断用户这句话是不是在问这个文档才交给模型。这一层还要做输出校验。模型返回的内容不能直接透传给用户或写入数据库必须过一遍校验。校验包括格式校验JSON 是否合法、内容校验是否包含敏感词、事实校验关键数字是否和检索结果一致。校验失败就走降级路径。4. 实操过程与核心环节实现4.1 从零搭建的最小可行架构假设现在要搭一个知识问答系统我按实际顺序走一遍。第一步是搭骨架目录结构这样组织app/ providers/ # 模型接入 base.py openai_provider.py local_provider.py workflow/ # 编排 states.py engine.py memory/ # 记忆 short_term.py long_term.py services/ # 业务逻辑 qa_service.py auth_service.py api/ # 交互层 routes.py这个结构的好处是每层职责单一测试时能单独 mock。我实测下来新人接手时理解成本比按功能分目录低很多。第二步是配置管理。AI 应用的配置项比传统应用多模型名、温度、超时、重试、向量库地址、top_k 等等。我建议用分层配置默认值写在代码里环境相关写在环境变量运行时可变的存在配置中心。温度这个参数问答场景我一般设 0.1 到 0.3创意生成才用 0.7 以上。4.2 一次完整请求的执行链路拿用户问上个月的销售数据是多少这个请求举例走一遍完整链路。请求进来先过鉴权确认用户有权限访问销售数据。然后进入编排层状态机从INTENT_PARSING开始。这一步调用模型判断意图提示词大概是判断用户问题的类型只输出以下之一data_query数据查询、knowledge_query知识问答、chitchat闲聊、unclear无法判断。 用户问题{question}模型返回data_query状态转移到数据查询分支。这里不直接让模型生成 SQL而是先用模型抽取查询参数时间范围、指标名再用确定性代码拼 SQL。为什么这么做因为让模型直接写 SQL 有注入风险而且字段名容易编造。抽取参数后代码校验参数合法性再执行查询。查询结果拿到后进入GENERATING状态把结构化数据交给模型生成自然语言回答。提示词里明确要求只使用提供的数据不要编造。生成完进入VALIDATING校验回答里的数字是否和查询结果一致。这一步用简单的字符串匹配加数值比对就能做。校验通过返回用户不通过走FALLBACK返回数据查询成功结果为 XXX这样的模板化回答。整条链路里模型只负责它擅长的部分理解意图、抽取参数、组织语言。确定性的部分全交给代码。这是我做 AI Native 项目最核心的一条经验。4.3 关键参数的计算与选择超时和重试参数前面提过这里说下 token 预算。假设模型上下文窗口是 128k但实际不该用满。我的分配是系统提示词 500 token检索结果 2000 token历史对话 1500 token用户输入 500 token留给输出的 2000 token。总共约 6500 token远低于窗口上限。为什么不把窗口用满两个原因。一是成本和延迟随 token 线性增长用满窗口既贵又慢。二是长上下文里模型对中间部分的注意力会下降这是实测能观察到的现象。所以我的原则是够用就好留足余量。向量检索的top_k和score_threshold需要根据数据调。我的调参方法是准备 50 条标注好的问答对跑一遍检索看正确答案的召回率和错误答案的引入率。top_k从 3 开始逐步加到 10找到召回率不再明显提升的点。score_threshold从 0.6 开始逐步加到 0.8找到错误率明显下降的点。这两个值没有通用答案必须用自己数据调。4.4 流式输出的实现要点流式输出对体验影响很大但实现上有几个坑。第一是首字节延迟用户等太久会以为卡死。我的做法是先发一个正在思考的信号再开始流式返回。第二是中途失败流到一半模型断了前端要能优雅处理。我的做法是前端维护一个缓冲区收到完整结束标记才渲染中途失败就回退到重试。第三是流式下的校验。非流式可以等全部生成完再校验流式就得边流边校验。我的做法是分段校验每积累一定长度就校验一次发现问题立即中断并降级。这样既保证安全又不牺牲体验。async def stream_with_validation(provider, messages): buffer async for chunk in provider.stream(messages): buffer chunk if len(buffer) 200: if not validate(buffer): yield [内容校验未通过已中断] return yield buffer buffer if buffer: yield buffer5. 常见问题与排查技巧实录5.1 模型输出格式不稳定怎么办这是最高频的问题。明明提示词里写了输出 JSON模型还是偶尔返回带 markdown 代码块的、带解释文字的。我的处理分三层提示词层用 few-shot 给例子接入层用response_format参数支持的模型代码层做容错解析。容错解析我写了个函数先尝试直接json.loads失败就正则提取第一个{...}块再失败就返回错误走降级。实测下来三层防护能把格式错误率从 5% 降到 0.5% 以下。注意不要指望提示词能 100% 解决格式问题模型本质是概率系统一定会有意外输出。工程上必须假设它会出错。5.2 响应慢怎么定位AI 应用的延迟来源比传统应用多。我一般按这个顺序排查先看模型调用耗时再看检索耗时最后看编排逻辑耗时。模型调用耗时又分首字节延迟和总生成时间前者反映网络和排队后者反映生成长度。我常用的排查表现象可能原因排查方法首字节延迟高网络或模型排队换区域、换模型测总生成时间长输出太长限制 max_tokens检索慢向量库索引未优化检查索引类型和分片整体慢但各环节正常编排串行过多改并行调用我踩过的一个坑是编排层里检索和意图判断串行执行其实这两个可以并行。改成并行后整体延迟降了 30% 多。5.3 成本失控怎么控制成本失控通常有三个原因上下文太长、重试太多、缓存缺失。上下文控制前面说了重试要设上限缓存是最容易被忽略的。缓存分两级精确缓存和语义缓存。精确缓存是相同输入直接返回上次结果用 Redis 就行。语义缓存是把输入向量化相似度超过阈值就复用结果。语义缓存能显著降低成本但要注意阈值不能太低否则会返回不相关的答案。我一般设 0.95 以上。5.4 多轮对话上下文丢失用户说那第二个呢系统不知道指什么。这是上下文管理没做好。我的解法是在每轮对话里维护一个指代消解步骤把第二个这类指代还原成具体对象再传给后续环节。具体做法是让模型做一次轻量的指代消解输入是最近几轮对话和当前问题输出是消解后的完整问题。这一步成本很低但能大幅提升多轮体验。5.5 模型幻觉怎么缓解幻觉是模型固有特性只能缓解不能消除。我的组合拳是检索增强让模型基于事实回答、输出校验比对关键信息、置信度评估让模型自评低置信度走人工、兜底话术不确定时明确说不知道。其中置信度评估有个技巧不要直接问模型你有多确定而是让它给出答案的同时列出依据依据是否充分比自评分数更可靠。6. 我在这几个项目里攒下的经验做 AI Native 架构这两年多最大的体会是架构的价值不在于用了多新的技术而在于把不确定性管理好。模型会变、提示词会调、数据会更新架构要能扛住这些变化。几个具体的经验。第一先跑通最小链路再优化。我见过团队一上来就设计复杂的多 Agent 协作结果连基本的问答都没跑通。先把输入—模型—输出这条线跑通再逐步加检索、加校验、加记忆。第二可观测性要早做。AI 应用的调试比传统应用难因为同样的输入可能得到不同输出。所以日志要记录完整的输入输出、模型参数、耗时、token 数。我一般用结构化日志方便后续分析。第三降级路径要设计好。模型服务会挂、会超时、会限流这时候系统不能整个不可用。我的做法是每个模型调用都有降级方案主模型挂了切备用模型备用也挂了走规则兜底规则兜底不了就明确告诉用户暂时无法处理。第四别过度依赖单一模型。不同模型在不同任务上表现差异很大有的擅长推理有的擅长生成有的便宜。接入层做好抽象后按任务路由到不同模型成本和效果都能优化。最后分享一个我常用的提示词结构模板实测下来比随意写的提示词稳定很多角色你是 XXX 任务完成 XXX 约束 1. 只使用提供的信息 2. 不确定时明确说明 3. 输出格式为 XXX 输入{input}这个结构把角色、任务、约束、输入分开模型更容易理解边界。约束部分尤其重要它决定了模型的行为边界。我一般会针对每个场景单独调约束比如数据查询场景加不要编造数字知识问答场景加引用来源。这套架构不是终点AI 领域变化太快今天的最佳实践明天可能就过时了。但底层的思路——隔离不确定性、显式编排、确定性代码兜底——我觉得会长期有效。把这些想清楚具体用什么框架、什么模型都是可以替换的细节。
返回列表