上个月我把一个知识库问答Agent从单一模型迁到多模型切换被各种分支判断搞得很狼狈。项目里同时要接三个模型来源一个商用模型API、一个公司内部微调的模型、一个本地部署的开源量化模型。每个来源的请求格式、超时行为、流式返回格式都不一样Agent主流程为了适配这些差异已经长出了一堆if/else再加新模型就得动主链路改起来处处是雷。后来花了一天时间把所有模型做了一次统一封装Agent主流程只依赖一个稳定的抽象接口换模型、加模型都变成了配置级操作。这篇记录的就是这次实践的全过程包括设计思路、核心代码以及我在流式解析和并发控制上踩过的坑。先说明一下这篇说的封装是软件层的模型接口封装不是PCB元器件封装或芯片封装那一类硬件概念。在Agent开发里自定义模型封装指的是把不同来源、不同协议、不同规格的大语言模型包装成一个统一的、对上层Agent友好的调用接口。它可以是项目里的一个Python包也可以是一个独立的模型网关服务。无论哪种形式目的都只有一个让Agent不需要关心底层模型是谁、在哪跑、怎么调用。这篇内容适合两类人。第一类是自己搭Agent、正被多模型接入搞到头大的开发者第二类是刚入门Agent开发、想搞明白自定义模型到底怎么接到LangGraph、CrewAI这类框架里的人。我会把关键步骤都展开尽量不讲空话。1. 先搞清楚模型封装到底在解决什么问题很多初学者把封装理解成写个函数把请求包一层这其实远远不够。如果只是不想重复写requests.post那你得到的充其量是一个工具函数不是封装。真正的隔离是要让上层业务逻辑完全不感知模型层面的差异。1.1 多个模型混在一起时真实代码会变得多乱我说一个实际场景。当时我的Agent要支持两个商用模型和一个本地模型代码里出现了类似这样的模式if model_name gpt-4o: # 调OpenAI格式 elif model_name minimax: # 调MiniMax格式参数名都不一样 elif model_name local: # 走本地推理还要自己做tokenize这只是请求入口。再往下走就是地狱A模型返回的错误码是invalid_request_errorB模型返回的是400加一段JSONC模型直接抛异常A模型流式返回按data:分割B模型是按choices[0].delta分割C模型的流式干脆是自定义协议。最后Agent的对话循环里到处是如果是这个模型就怎么处理如果是那个模型就怎么处理。我把这种状态叫做模型差异污染了业务逻辑。它的后果不是丑而是改不动想加一个新模型你得改Agent主流程、改错误处理、改流式解析、改token统计改完还不敢保证不影响其他模型。项目里的人越多这里就越像雷区。1.2 封装之后你收获了什么把这三个模型统一封装之后Agent主流程的调用代码变成了这样response await provider.chat(request)或者流式场景async for chunk in provider.chat_stream(request): handle(chunk)就这么干净。底层是哪个模型、走什么协议、怎么处理超时Agent一概不管。我总结下来封装带来的收益主要有四个。第一是标准契约。所有的模型都遵循同一个抽象接口入参、出参、异常、流式格式全部统一上层不需要关心实现。第二是可替换性。今天用商用模型明天换本地模型Agent主流程一个字都不用改。第三是可观测性。因为所有请求都走同一个入口日志、耗时、token统计、错误率全部可以在这个入口统一采集。第四是容错能力。超时重试、降级、并发限制这些横切逻辑可以集中在一个地方做不用在每个模型适配器里重复实现。1.3 Agent逻辑和模型提供方之间的边界这里要稍微说清楚一个边界问题。Agent本身是一套决策-行动-观察的循环它负责理解目标、拆分步骤、调用工具、根据结果继续决策。模型在这套循环里扮演的是推理引擎它接收上下文输出下一步该怎么做的判断。而不同来源的模型推理能力、上下文长度、输出格式都不相同。如果Agent主流程直接依赖某个具体模型的SDK那Agent就和这个模型绑死了。我见过不少项目Agent里所有Prompt和工具调用的逻辑都写在一个文件里然后通过import openai直接调API看起来简单但模型一换整份代码都要跟着改。把模型提供方单独抽象出来本质上是在Agent和具体模型之间加了一道翻译层Agent说我要聊天、我要流式输出、我要知道token消耗翻译层负责把这句话转换成具体模型的协议。这一步想明白了后面所有的设计都顺理成章。2. 封装设计的核心思路定一套协议而不是包一层函数既然目标是隔离那设计的起点就不是某个具体模型的SDK而是我们的Agent到底需要模型提供什么能力。把这个能力清单定下来就是抽象接口的原型。2.1 用抽象基类定义模型提供方的契约我在项目里定义基类时参考了Python里常见的接口先于实现做法。基础接口只保留四个核心能力普通对话、流式对话、返回token使用量、标识当前提供方名称。大致长这样from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import AsyncIterator, Optional dataclass class ChatMessage: role: str # system / user / assistant content: str dataclass class ChatRequest: messages: list[ChatMessage] model: str temperature: float 0.7 max_tokens: int 2048 stream: bool False extra: dict field(default_factorydict) dataclass class TokenUsage: prompt_tokens: int completion_tokens: int total_tokens: int 0 def __post_init__(self): if self.total_tokens 0: self.total_tokens self.prompt_tokens self.completion_tokens dataclass class ChatResponse: message: ChatMessage usage: TokenUsage model: str raw: Optional[dict] None class BaseModelProvider(ABC): property abstractmethod def provider_name(self) - str: 返回模型提供方名称用于日志和监控。 abstractmethod async def chat(self, request: ChatRequest) - ChatResponse: 非流式对话返回完整响应。 abstractmethod def chat_stream(self, request: ChatRequest) - AsyncIterator[str]: 流式对话逐块产出文本增量。看到没有这里刻意把流式定义为AsyncIterator[str]而不是返回一个完整字符串。这样做的原因是流式场景下调用方想做什么都很自由可以边收边渲染可以攒起来手动拼接也可以在收到某个特殊标记时提前中断。如果封装层把流式结果拼好再返回就丧失了流式的意义。2.2 接入方式选型OpenAI兼容协议优先还是自定义协议优先定好接口之后第二个关键决策就来了各个模型到底怎么接进来。这块我见过两种倾向。一种是花钱省事派所有模型都强制走OpenAI兼容协议因为现在大量商用API和开源推理框架vLLM、TGI、Ollama都提供OpenAI兼容端点。另一种是自力更生派觉得OpenAI协议不够灵活要为每个模型写原生的HTTP调用和解析。我的建议是能用OpenAI兼容协议就优先用只有兼容协议覆盖不了的时候才写自定义适配器。原因很简单OpenAI兼容协议是目前事实上的行业标准社区工具、可观测性插件、Agent框架对它的支持最成熟。你只要把自己的封装对外暴露成OpenAI格式哪怕未来换了一个框架也能无缝对接。下面是这两种选型的对比我实际用下来的感受都在表里选型优点缺点适用场景OpenAI兼容协议生态成熟、工具链多、对接成本低部分特有参数要映射可能有表达不了的功能商用API、vLLM/Ollama部署的开源模型、公司内统一网关自定义协议参数表达自由、能保留模型全部能力解析和测试成本高、生态兼容差、后期维护重自研模型服务、内部私有协议、无现成OpenAI端点这里有一个小经验当某个模型提供了OpenAI兼容端点但文档里又额外扩展了参数时不要急着为扩展参数写死逻辑。把那些参数塞进ChatRequest.extra字典里透传下去比在基类里每个参数都定义一个字段要灵活得多。基类字段只保留几乎所有模型都通用的temperature、max_tokens、stream就够了。2.3 流式输出怎么封装才不膈应人流式是Agent开发里的重头戏也是封装最容易出问题的地方。Agent场景下流式通常有两种用途一种是把模型的增量输出实时推给用户界面另一种是Agent在思考过程中边生成中间内容边执行工具调用。无论哪种封装层都要保证三点。第一不能吞掉流式块。模型返回多少块、每块什么内容封装层尽量原样透传给调用方最多做格式转换不要自作主张拼接或丢弃。第二必须能提前取消。用户可能随时打断Agent的回答如果封装层不支持在迭代途中退出那中断一个长任务就会变成灾难。第三异常要能在迭代过程中抛出。流的中间可能断掉、返回错误码、或者干脆没有内容异常不能攒到最后才让人看到。实现流式适配器时我习惯把SSE的解析逻辑单独抽出来。SSEServer-Sent Events是大多数模型服务端流式响应的底层格式每一行以data:开头以空行分隔事件。解析时最忌讳的是按json.loads去解析整段响应因为流式数据是一块一块发过来的一个事件可能被拆成两半。正确的做法是按行读取遇到data:开头的行就尝试解析解析失败就继续累积直到凑成完整JSON。2.4 服务地址和模型名千万别写死在代码里封装设计里还有一个容易被忽略的点配置化。服务的api_base、模型名、密钥这些一定不要让每个适配器自己去硬编码。我见过有的项目把服务地址直接写在适配器构造函数里结果换测试环境和生产环境时要改代码这是很伤的。正确的做法是让适配器从配置中心、环境变量或配置文件中读取。这样一来自定义模型就变成了填一个服务地址、填一个模型名、填一个密钥的配置动作。市面上很多工具所谓的自定义模型服务地址本质上就是同一个套路它定义好一套协议你填一个地址让它去访问只要你的服务响应格式符合预期模型就接进来了。这不是什么黑魔法就是把协议标准化之后让接入动作退化成配置。3. 实操落地手写一套可用的自定义模型封装理论说多了没用下面进入实操。我拿一个真实项目里的简化版本来做讲解代码可以当成脚手架直接抄。演示用的模型来源有两个一个走OpenAI兼容协议一个走本地Transformers推理。3.1 项目目录与数据契约我的目录结构是这样的model_provider/ ├── __init__.py ├── base.py # 抽象基类、数据类 ├── openai_compat.py # OpenAI兼容协议适配器 ├── local_hf.py # 本地HuggingFace模型适配器 ├── registry.py # 提供方注册与工厂 └── config.py # 配置读取base.py里就是上一节展示的那几个类。实际项目里我还会加一个ProviderError异常类用来统一不同模型返回的错误信息字段包括provider_name、status_code、message。这样Agent捕获异常时不需要解析每个模型的原始错误结构拿到的一定是统一格式。3.2 OpenAI兼容模型适配器实现这个适配器覆盖了绝大多数场景你有一台部署了vLLM的GPU服务器或者你在用一个提供OpenAI兼容端点的商用模型下面的代码都能直接工作。import asyncio import json from typing import AsyncIterator, Optional import httpx from .base import ( BaseModelProvider, ChatRequest, ChatResponse, ChatMessage, TokenUsage, ) class OpenAICompatProvider(BaseModelProvider): def __init__( self, name: str, api_base: str, api_key: str, model: str, timeout: float 60.0, max_concurrency: int 8, ): self.name name self.api_base api_base.rstrip(/) self.api_key api_key self.model model self.timeout timeout self._semaphore asyncio.Semaphore(max_concurrency) property def provider_name(self) - str: return self.name def _url(self) - str: return f{self.api_base}/chat/completions def _headers(self) - dict: return { Authorization: fBearer {self.api_key}, Content-Type: application/json, } async def chat(self, request: ChatRequest) - ChatResponse: payload { model: request.model or self.model, messages: [m.__dict__ for m in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: False, } if request.extra: payload.update(request.extra) async with self._semaphore: async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.post(self._url(), jsonpayload, headersself._headers()) resp.raise_for_status() data resp.json() usage data.get(usage, {}) return ChatResponse( messageChatMessage( roleassistant, contentdata[choices][0][message][content], ), usageTokenUsage( prompt_tokensusage.get(prompt_tokens, 0), completion_tokensusage.get(completion_tokens, 0), ), modeldata.get(model, payload[model]), rawdata, ) async def chat_stream(self, request: ChatRequest) - AsyncIterator[str]: payload { model: request.model or self.model, messages: [m.__dict__ for m in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: True, } if request.extra: payload.update(request.extra) async with self._semaphore: async with httpx.AsyncClient(timeoutself.timeout) as client: async with client.stream( POST, self._url(), jsonpayload, headersself._headers() ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line.startswith(data:): continue data_str line[5:].strip() if data_str [DONE]: return try: obj json.loads(data_str) except json.JSONDecodeError: continue choices obj.get(choices, []) if not choices: continue delta choices[0].get(delta, {}) content delta.get(content, ) if content: yield content这里我刻意用了httpx.AsyncClient因为它在异步流式读取上比requests舒服得多而且超时控制、连接复用表现都更好。_semaphore用来限制最大并发数避免Agent并发调用时直接把模型服务打崩。关于并发问题后面第4节会展开讲。你可能会问为什么要用AsyncIterator[str]一层层yield而不是直接返回一个拼接好的字符串因为Agent在流式场景下往往需要在模型生成的同时做处理。比如用户在终端看到了逐字输出的效果或者流式内容里出现了工具调用的关键字Agent要立即响应。这些场景都需要边收边处理的能力。3.3 本地模型适配器实现很多团队的私有数据场景不允许把请求发到外部服务必须用本地模型。本地模型适配器的核心问题不是HTTP调用而是要把一个同步的、可能很耗时的推理过程放到异步接口里跑。import asyncio from typing import AsyncIterator, Optional from threading import Lock from .base import ( BaseModelProvider, ChatRequest, ChatResponse, ChatMessage, TokenUsage, ) class LocalHFProvider(BaseModelProvider): def __init__(self, model_path: str, device: str cpu, max_batch: int 4): self.model_path model_path self.device device self._max_batch max_batch self._tokenizer None self._model None self._lock Lock() self._load_model() def _load_model(self): # 实际项目中可以换成AutoTokenizer / AutoModelForCausalLM # 这里用占位说明避免引入过重的依赖 pass property def provider_name(self) - str: return flocal-{self.model_path} async def chat(self, request: ChatRequest) - ChatResponse: # 把同步推理丢进线程池防止阻塞事件循环 return await asyncio.get_running_loop().run_in_executor( None, self._sync_generate, request ) def _sync_generate(self, request: ChatRequest) - ChatResponse: with self._lock: prompt self._build_prompt(request.messages) output self._infer(prompt) return ChatResponse( messageChatMessage(roleassistant, contentoutput), usageself._estimate_usage(prompt, output), modelself.model_path, ) async def chat_stream(self, request: ChatRequest) - AsyncIterator[str]: # 先走非流式推理再按token粒度切分输出 response await self.chat(request) content response.message.content for i in range(0, len(content), 4): yield content[i:i 4] def _build_prompt(self, messages): return \n.join(f{m.role}: {m.content} for m in messages) def _infer(self, prompt: str) - str: # 这里放实际模型推理代码 return local generated text def _estimate_usage(self, prompt: str, output: str) - TokenUsage: # 本地模型没有usage字段用字符数粗略估算 prompt_tokens len(prompt) // 3 1 completion_tokens len(output) // 3 1 return TokenUsage( prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, )这个适配器有两个关键点。第一本地模型的推理是阻塞的直接放进协程里会卡死整个事件循环必须用run_in_executor丢到后台线程。第二本地模型往往没有标准的token统计但Agent的上层逻辑可能依赖usage来做预算控制所以我在适配器里做了估算。估算公式先按字符数/3算个大概够用真要精确可以接tokenizer去做count。3.4 把封装组件挂到Agent主流程有了适配器之后怎么让Agent用上这套封装我推荐做一个Registry让Agent不问来路只按名字取Providerclass ProviderRegistry: def __init__(self): self._providers {} def register(self, provider: BaseModelProvider): self._providers[provider.provider_name] provider def get(self, name: str) - BaseModelProvider: if name not in self._providers: raise KeyError(fprovider {name} not registered) return self._providers[name]Agent主流程里的用法registry ProviderRegistry() registry.register(OpenAICompatProvider( namegpt-4o-mini, api_baseos.environ[OPENAI_BASE_URL], api_keyos.environ[OPENAI_API_KEY], modelgpt-4o-mini, )) registry.register(OpenAICompatProvider( namevllm-7b, api_baseos.environ[VLLM_BASE_URL], api_keyEMPTY, modelmy-model-7b, )) provider registry.get(vllm-7b) response await provider.chat(ChatRequest(messages[...]))这里可以看出封装的效果Agent要切换模型只是改一行registry.get的入参。如果以后要加一个支持自定义协议的新模型只需要再写一个适配器并注册进去原来所有依赖BaseModelProvider的代码都不受影响。3.5 更进一步把封装独立成模型网关如果你的团队有多个项目都要复用这套模型接入能力我建议把封装推进一步升级成一个独立的小服务对外暴露OpenAI兼容的接口。这一步的工作量不大但收益巨大。核心逻辑是用FastAPI包一层把收到的请求转换成我们的ChatRequest再转发给对应Provider。下面是一个最小可用的伪代码from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import json app FastAPI() app.post(/v1/chat/completions) async def chat_completions(request: Request): payload await request.json() provider_name payload.pop(provider, default) provider registry.get(provider_name) req ChatRequest( messages[ChatMessage(**m) for m in payload[messages]], temperaturepayload.get(temperature, 0.7), max_tokenspayload.get(max_tokens, 2048), streampayload.get(stream, False), extrapayload, ) if not req.stream: resp await provider.chat(req) return { choices: [{message: {role: assistant, content: resp.message.content}}], usage: {prompt_tokens: resp.usage.prompt_tokens, completion_tokens: resp.usage.completion_tokens} } else: async def event_gen(): async for chunk in provider.chat_stream(req): data {choices: [{delta: {content: chunk}}]} yield fdata: {json.dumps(data)}\n\n yield data: [DONE]\n\n return StreamingResponse(event_gen(), media_typetext/event-stream)这一步的意义在于服务一旦暴露成标准OpenAI格式LangFlow、VSCode、各种Agent框架都可以通过自定义模型服务地址直接接入不需要知道你的模型是本地还是商用、是微调的还是开源的。我在项目里完成这一步之后团队的算法同学自己就能在可视化工具里测新模型不用再排队等开发改代码。4. 真实项目中的排查记录常见问题与避坑封装完成只是开始真正让我长记性的是后面这一段。我把实际遇到的问题整理成了一份排查清单按出现频率从高到低排序。4.1 高频问题速查表症状可能原因处理方式流式输出只出几行就卡住超时时间设置太短长回答生成超过预期把timeout放宽到120秒以上或者做成可配置流式内容随机缺失解析时按行判断但没处理省略号开头的行加上对data:前缀的严格过滤同时处理无前缀行并发一高就大量超时没有做并发限制模型服务被打满适配器里加信号量同时调大连接池token统计不准某些模型不返回usage字段在适配器内做估算兜底换模型后Prompt效果变差不同模型指令遵循能力不同保留每个模型独立的系统提示词版本本地模型推理卡死了整个服务同步推理阻塞了事件循环用线程池隔离限制最大任务数4.2 SSE流式响应解析失败的完整排查我在第一次写OpenAI兼容适配器时遇到过一个很诡异的现象流式输出偶尔少一段而且不是固定位置少。排查了半天最后定位到是SSE解析的问题。标准SSE格式里事件之间用空行分隔每个字段以字段名:开头。回到3.2的代码我只处理了以data:开头的行。但实际传输中如果网络包边界刚好把一行切开了aiter_lines拿到的行也可能是不完整的。我第一次用的解析逻辑是拿到一行就立刻json.loads结果遇到半行数据就解析失败直接continue跳过了这一块内容就丢了。修复方案是把解析逻辑改成缓冲-重试模式开一个临时列表累积片段只有拼成完整JSON时才继续如果拼完还是失败再显式抛出异常。这种解析逻辑我建议单独拆成一个工具函数给所有流式适配器共用不要每个适配器各写一遍。另一个坑是流式结束标记。OpenAI协议的结束标记是data: [DONE]有的服务端会在这个标记后面再加一个空行有的不会。判断结束条件时一定先取line[5:].strip()再判断否则被空白字符干扰就会漏判导致流式迭代无限等下去。4.3 并发一上来就超时的问题Agent框架的特性是常常会并行调用多个模型来完成子任务。如果Agent内部有5个分支同时调用同一个模型而模型服务本身只支持少量并发那超时几乎是必然的。我踩过的一次实际经历本地vLLM服务部署了一个13B模型OpenAI兼容端点看着很正常但Agent一跑多路并发马上出现大量timeout和ConnectionError。一开始我以为是服务端配置问题后来才发现是我自己的适配器没有做并发限制一次放了20个请求过去把服务挤爆了。解法就是我在适配器里加的那个Semaphore。信号量的值不要拍脑袋定最好根据模型服务的实际情况来设。vLLM部署时如果你设置了--max-num-seqs那信号量上限就不要超过这个数商用API则可以参考它文档里给的RPM/TPM限制先给一个保守值再逐步压测调整。还有一个容易忽略的细节httpx.AsyncClient如果每次调用都新建实例连接池就无法复用高频请求会频繁建立TCP连接带来不必要的延迟。我建议把AsyncClient作为适配器实例的一个复用成员或者至少用httpx.Client的全局连接池让连接能保持复用。4.4 token统计不准与上下文长度控制Agent框架普遍会遇到上下文过长问题而token统计是解决它的前提。我在开发中遇到的情况是三家模型返回的usage结构完全不一样有的字段叫prompt_tokens有的叫input_tokens有的压根不返回。如果适配器不统一Agent那边写上下文裁剪逻辑时会非常痛苦。我的统一方案是在适配器里做“兜底转换”能取到官方usage就转换进我们的TokenUsage取不到就用字符数估算。这个方案不完美但足够让Agent层的上下文管理逻辑先稳定运行。上下文长度控制还有一个常见误区只看max_tokens参数忽略了模型真实的上下文窗口。比如模型窗口是8K你max_tokens设了6K那输入最多只能有2K。如果Agent塞进去的历史记录太多首包返回就直接报错或丢历史。我后来在适配器里做了一层防护请求发出之前用tokenizer或简单估算接口粗算一下输入长度超过阈值自动截断最旧的历史消息。这层逻辑放在适配器里比放在Agent主流程里更合适因为只有适配器最清楚自己接的模型到底允许多长输入。5. 给后来者的一些实际经验代码部分讲完了最后聊几个偏软的经验是我在这套封装上迭代了几版之后才想明白的。5.1 封装别过度设计最小契约跑通再扩展第一次做封装的人容易陷入一个误区想在一开始就把所有能力做全包括自动重试、熔断、缓存、负载均衡、全链路追踪。我建议不要这样。第一版只要保证两件事能跑非流式对话、流式对话。等跑通之后再往里加能力。为什么因为很多能力你以为是通用的加进去才发现不同模型的行为差异很大。比如重试策略有些模型服务对重复请求的幂等处理很弱你无脑重试反而会造成重复消费。比如缓存流式输出要不要缓存按什么粒度缓存这些都需要实际场景验证设计阶段拍脑袋做的方案大概率要返工。5.2 日志和可观测性是一等公民封装层是所有模型请求的必经之路这里是做可观测性最好的位置没有之一。我在适配器里统一打印了一条结构化日志大概包含这些字段时间、Provider名称、目标模型名请求ID方便和Agent主流程关联输入token数、输出token数、总耗时是否流式、HTTP状态码如果有错误类型和错误消息如果有有了这条日志线上出问题时排查非常快。有一次用户反映某个模型回复突然变慢我直接查日志发现P99耗时从1.2秒涨到了8秒再结合Provider名称定位到是本地模型所在机器负载高了。如果没有统一日志层这种问题要翻几个系统的原始日志才能拼出全貌。5.3 可以继续扩展的方向这套封装跑稳之后后续有几个很自然的扩展方向。第一个是Agent记忆模型层的TokenUsage数据可以喂给上下文管理模块按预算自动压缩记忆第二个是模型路由在Registry之上再加一层路由策略根据任务类型或成本预算自动选择Provider第三个是模型网关服务化之后加认证和限流让不同团队共享同一个模型入口而不互相干扰。这几个方向里我目前觉得收益最大的是模型路由。以前所有请求都打到一个模型上成本和延迟都很高做了路由之后简单问题走快模型复杂推理走慢模型整体成本和响应速度都有明显改善。回到开头那个多模型切换的场景我最大的体会是封装这件事表面上省的是代码量实际上是省掉了换模型时的恐惧感。模型迭代速度很快今天还在用的API明天可能就涨价、降级甚至下线如果没有一个稳定的模型接入层每一次模型变更都会变成一次牵一发动全身的重构那Agent项目的迭代节奏根本快不起来。如果你正在做Agent项目建议从最小契约开始花半天时间把模型层隔离出来。等基础跑通再逐步把流式、并发、可观测性这些能力补进去。这件事越早做后面的收益越大。