这次我们不看一个具体的开源项目，而是拆一个更值得关注的行业现象：70% of AI revenue comes from OpenAI and Anthropic。翻译成大白话就是，AI 行业里的真金白银，大部分流向了 OpenAI 和 Anthropic 这两家头部闭源模型公司。对做技术的同学来说，这从来不是新闻稿，它直接影响我们的技术选型、API 预算、模型采购策略，甚至决定我们该不该花时间学一套新的 AI 开发框架。
标题这句话本身带有一些统计口径的争议，不同咨询机构的数字也不完全一致，但大方向很清楚：AI 收入高度集中，二八效应变成了“二七”效应。这个现象背后不只是产品做得好，更多是技术能力、API 生态、开发者工具、企业服务、算力和投资共同叠加的结果。这篇文章我从工程师视角拆一下：为什么收入会集中到这两家，OpenAI 和 Anthropic 的 API 生态到底强在哪，开发者怎么用兼容层统一接入两家服务，以及面对“闭源 API 涨价 + 数据不出域”的约束，本地开源模型和批量任务成本控制应该怎么做。
文章后面会给出可以直接上手的部署思路、Python 调用示例、批量任务队列模板、显存和连接排障清单。如果你想做 AI 应用，又不想被单一厂商绑死，这篇建议收藏。
1. 核心现象速览
在展开技术细节之前，先用一张表把当前 AI 收入集中现象和开发者能切入的路径梳理清楚。
| 维度 | 现状说明 |
|---|---|
| 收入集中方 | OpenAI、Anthropic 两家占据 AI 行业大部分收入，尤其在企业级服务和 API 调用侧 |
| 主要产品形态 | ChatGPT、Claude 等 C 端产品，以及面向开发者的 OpenAI API、Anthropic API |
| 开发者主要切入方式 | 直接调用官方 API、通过兼容层统一接入、在开源模型私有化部署后自建服务 |
| 生态壁垒来源 | 模型能力领先、API SDK 完善、插件与智能体生态完整、企业合规服务成熟 |
| 开源替代路线 | 可用本地推理框架部署开源模型，成本可控但效果和易用性需要实测评估 |
| 开发风险点 | API 成本不可控、单一厂商依赖、数据隐私、接口变更、模型下架 |
| 应对策略 | 多模型路由、兼容层封装、预算熔断、本地兜底、异步批量任务 |
这里要解释清楚一个容易误解的点：AI 收入不等于 AI 投入。从 2023 年到 2025 年，AI 领域的融资大头也集中在头部模型公司，但文章里提到的“70% 收入集中”更多指的是模型调用费、订阅费、企业授权费这些实际营收，而不是单纯的资本投入。对开发者来说，我们关注的是实际发生成本的 API 调用侧，这才是每天要跟预算、限流和账单打交道的地方。
从技术生态看，OpenAI 和 Anthropic 之所以能同时吃下收入和开发者心智，有几个共同特征：模型效果好，官方 SDK 支持语言多，文档更新快，生态工具链完整，同时在企业服务上提供私有化或审计合规方案。后面几个章节我会逐个拆。
2. 为什么 AI 收入会集中到 OpenAI 和 Anthropic
2.1 技术能力决定了“贵也有人买”
企业付费和开发者选择模型，首先看的还是效果。OpenAI 和 Anthropic 的旗舰模型在长文本、代码生成、复杂推理、工具调用这些关键场景上长期处于第一梯队。企业做 AI 客服、代码助手、数据分析这类产品时，模型效果直接决定产品能不能上线。当闭源模型效果领先，API 定价就有溢价空间，企业为了不折腾自研，反而更愿意付费。
2.2 API 生态和开发者体验形成强粘性
OpenAI 的 API 设计得非常规范，/v1/chat/completions接口被大量第三方框架当作事实标准。Anthropic 的 API 在 Messages API 设计上更强调system提示词和多轮消息结构，也提供了大量安全调节参数。两者都有完整 SDK，Python、Node.js、TypeScript 都有官方支持。这种“开箱即用”让开发者的迁移成本非常高，一旦项目里大量代码依赖某个 SDK，换模型的成本不光是改一个 base_url，还可能涉及工具调用格式、流式协议、返回字段差异的适配。
2.3 企业级服务能力拉高了客单价
个人开发者可能更关注 API 价格和模型效果，但企业采购更关注稳定性和合规。OpenAI 和 Anthropic 都提供企业版服务，有更细粒度的权限管理、审计日志、数据保留策略，并且支持不将企业数据用于模型训练。对金融机构、医疗、法律这些强监管行业来说，这个能力是刚需。企业一旦签订年度合同，收入就变成可预期的营收，这就是收入的“压舱石”。
2.4 算力和芯片布局拉大长期差距
从公开信息看，OpenAI 在芯片自研上动作很快，有团队在推进自研 AI 芯片，目标是在训练和推理成本上摆脱对单一厂商的依赖。自研芯片的好处在于：长期可控的算力成本、更匹配自家模型的架构优化、更高的训练吞吐。头部模型公司一旦在算力侧获得成本优势，模型价格就能压得更低，中小模型公司很难跟进。Anthropic 则和云厂商绑定很深，官方 API 的可用性也很稳定，让企业放心把生产流量挂上去。
2.5 开源工具反哺平台生态
OpenAI 在 2025 年前后逐步开源了 Codex 引擎相关的工程组件，包括 CLI 工具和用于评估编码智能体的沙箱环境。这意味着原本在 ChatGPT 内部使用的 AI 编程闭环，开发者也可以自己搭：模型调用、代码执行、报错反馈、自动修复。这类“开源工具 + 闭源模型”的组合形成了对开发者的包围：工具是免费的，但真正跑起来的推理成本要付给平台。生态越来越大，收入自然集中。
这里补充一点判断：收入集中短期内不会消失，但不会无限恶化。开源模型和本地部署方案在某些垂直场景里成本优势明显，后面会单独讲。
3. 对开发者和技术团队的影响
收入集中在头部两家，对做技术的人影响是实打实的，不是宏观报告里的数字。
3.1 单一 API 依赖风险
很多团队现在把核心业务直接构建在某个平台 API 上。一旦平台调整模型价格、修改接口语义、或者临时限流，业务就会被波及。比如高峰期 API 返回 429 错误，用户侧表现为“AI 突然不说话了”，排查半天才发现是上游限流。
3.2 成本结构不确定
LLM API 是“用多少付多少”的模式，不像服务器按月付费。上线一个 AI 产品后，随着用户量增长，API 账单可能翻几倍。如果没有预算监控，月底账单会让你印象很深刻。更麻烦的是，很多平台的计费还区分输入 token、输出 token、缓存 token、图片 token，成本模型比普通 Web 服务复杂一个量级。
3.3 数据隐私与合规
调用外部 API 意味着业务数据要离开自己的服务器。涉及用户隐私、客户商业机密时，合规部门不会同意。这时候要么选企业版协议里的数据隔离条款，要么直接本地部署开源模型，但本地部署又面临效果和运维成本的问题。
3.4 技术栈锁定
官方 SDK 虽然好用，但它会把你的代码慢慢长成“某个平台的形状”。以后想换成别的模型厂商，一堆代码要重写。所以现在很多技术团队会在一开始就引入兼容层或网关，把模型厂商隔离在业务代码外面。
这几点不是让大家放弃 OpenAI 和 Anthropic，而是提醒：用闭源大模型的同时，要有“随时能换”和“随时能降级”的工程预案。
4. 技术选型：闭源 API 与本地开源模型怎么选
聊完现象，进到实操层面。最核心的问题：业务到底应该用闭源 API，还是本地部署开源模型。这个没有标准答案，先看对比表。
| 对比维度 | 闭源 API（OpenAI / Anthropic） | 本地开源模型 |
|---|---|---|
| 效果天花板 | 高，旗舰模型持续迭代 | 取决于所选模型版本，一般弱于闭源旗舰 |
| 接入成本 | 低，注册拿 Key 就能调 | 高，需要 GPU 服务器、模型文件、推理框架 |
| 数据控制 | 看企业协议，默认数据要出域 | 完全本地化，数据不出服务器 |
| 运维成本 | 基本为零，平台管稳定性 | 要管显存、日志、版本、模型更新、并发 |
| 成本结构 | 按 token 付费，用量大了很贵 | 前期硬件成本高，后期边际成本低 |
| 并发弹性 | 平台自动扩容，有配额限制 | 需要自己做推理服务扩容 |
| 适用场景 | 原型验证、效果优先、快速上线 | 数据敏感、调用量极大、需要定制模型 |
我的建议是混合路线：
- 原型阶段用闭源 API，快速验证产品价值。
- 数据敏感场景用本地开源模型，哪怕效果差一点，也要先保证合规。
- 高频、低价值、模板化任务用本地小模型或规则兜底，把大模型 API 留给高价值任务。
- 用兼容层把模型提供商抽象成配置项，随时可以切换。
这条路线的核心工程组件就是兼容层和网关，下一节讲。
5. 用兼容层统一接 OpenAI 与 Anthropic 接口
5.1 兼容层解决什么问题
不同模型厂商的 API 格式不统一，OpenAI 是/chat/completions格式，Anthropic 是/v1/messages格式，返回字段名、工具调用格式、流式事件名都不一样。兼容层的作用就是提供一个统一接口，内部再转换成各家 API 的请求格式。
这样业务代码只需要对接一套接口，底层用 OpenAI 还是 Anthropic，只是配置项的差异。目前常见的方案有 LiteLLM、one-api 这类网关项目，可以根据项目情况选型。
5.2 LiteLLM 基础用法
LiteLLM 是一个很流行的 Python 库和代理服务，可以统一调用多个大模型平台。安装和调用示例：
pip install litellmimport litellm import os # 环境变量注入 Key，不要硬编码在代码里 os.environ["OPENAI_API_KEY"] = "sk-xxxx" os.environ["ANTHROPIC_API_KEY"] = "sk-ant-xxxx" response = litellm.completion( model="anthropic/claude-3-5-sonnet-20241022", messages=[ {"role": "user", "content": "用三句话解释一下什么是 AI Agent 的规划模块"} ], temperature=0.2, ) print(response["choices"][0]["message"]["content"])注意上面的model参数实际格式要按 LiteLLM 文档查看，通常是厂商前缀/模型名的格式。它底层会判断需要调 OpenAI 还是 Anthropic，并把响应统一成 OpenAI 风格的 dict，业务代码处理起来就方便很多。
5.3 同时配置多个模型和自动降级
更实用的是让同一个业务请求可以自动降级：主模型用 Anthropic，请求失败或超时自动切到 OpenAI，再不行切到本地模型。这样可以显著提高服务的可用性。下面是一个简化的配置思路：
# 假设使用 LiteLLM 的代理服务配置 model_list: - model_name: primary litellm_params: model: anthropic/claude-3-5-sonnet-20241022 api_key: os.environ["ANTHROPIC_API_KEY"] - model_name: fallback litellm_params: model: openai/gpt-4o api_key: os.environ["OPENAI_API_KEY"]实际配置字段可能随版本更新而调整，部署时建议先看官方文档。核心思想是：业务代码永远只感知一个模型名称，模型路由、降级、重试全部交给代理层处理。
5.4 直接调用两个官方 SDK 的最小示例
如果你的项目暂时不需要代理层，也可以直接写两个官方 SDK 的最小代码，方便理解两种 API 的差异：
# OpenAI SDK 调用示例 from openai import OpenAI client = OpenAI() response = client.chat.completions.create( model="gpt-4o", messages=[ {"role": "user", "content": "用一句话解释 KV Cache"} ], temperature=0.2, ) print(response.choices[0].message.content)# Anthropic SDK 调用示例 from anthropic import Anthropic client = Anthropic() response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[ {"role": "user", "content": "用一句话解释 KV Cache"} ], ) print(response.content[0].text)两个 SDK 的差异点很直观：
- OpenAI 返回对象里的内容是
choices[0].message.content。 - Anthropic 返回对象里的内容是
content[0].text，而且必传max_tokens。 - Anthropic 的消息格式中
system不是普通 user 消息，而是单独字段。
如果代码里同时兼容两种 SDK，每个调用的地方都要处理这些差异，维护成本很高。所以生产环境里更推荐引一层兼容。
6. 批量任务与成本控制实践
6.1 批量任务为什么难
大模型 API 处理批量任务时，常见的问题有三个：
- 并发限制，平台的 RPM 和 TPM 配额有限。
- 单个任务耗时长，可能是秒级到分钟级。
- 失败率不低，网络超时、限流、上下文超限都可能出现。
所以批量任务一定要有队列、并发控制、重试、日志和结果落盘。最简单的做法是用 Python 的threading配合队列，但在生产环境更推荐用 Celery、Arq 这类异步任务框架。
6.2 一个带重试的批量调用示例
下面是一个简化但完整的批量任务模板，演示了如何用队列 + 线程池并发调用 Anthropic API，带失败重试和结果落盘。
import os import threading import queue import time from concurrent.futures import ThreadPoolExecutor from anthropic import Anthropic ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"] MODEL_NAME = "claude-3-5-sonnet-20241022" client = Anthropic() def process_one(text: str, max_retries: int = 3) -> dict: for attempt in range(max_retries): try: resp = client.messages.create( model=MODEL_NAME, max_tokens=512, messages=[{"role": "user", "content": text}], timeout=60, ) return {"ok": True, "text": resp.content[0].text} except Exception as e: print(f"[retry {attempt + 1}] failed: {e}") time.sleep(2 ** attempt) return {"ok": False, "text": ""} def worker(q: queue.Queue, results: list): while True: item = q.get() if item is None: break idx, text = item res = process_one(text) results.append((idx, res)) q.task_done() if __name__ == "__main__": tasks = [ "总结这段内容", "翻译这段内容", "提取关键信息", ] q = queue.Queue() results = [] for i, t in enumerate(tasks): q.put((i, t)) # 控制并发数，避免触发平台限流 threads = [] with ThreadPoolExecutor(max_workers=2) as executor: for _ in range(2): executor.submit(worker, q, results) q.join() for row in results: print(row)这个示例里我故意没把队列关闭逻辑写得很复杂，实际项目建议用现成任务框架，但核心原则是通用的：任务进队列、并发受限、失败指数退避重试、结果单独收集。
6.3 成本控制的工程手段
- 对用户输入做长度截断，超长文本先做摘要再走大模型。
- 模板化任务改用本地小模型，只有关键步骤调用大模型。
- 相同的用户问题和系统提示词，加一层缓存，命中缓存就不调用 API。
- 对单用户、单任务的调用频率和 token 上限做配额，防止被恶意刷量。
- 设置每日/每月的预算线，预算到阈值自动熔断，改用备用模型或返回降级提示。
成本控制的目的不是把 API 调用压到最低，而是让每笔 token 消费都有明确业务收益。
7. 开源模型本地部署的替代路径
7.1 本地推理框架怎么选
如果决心做本地部署，可以使用 Ollama、vLLM、Text Generation Inference 这类推理框架。Ollama 适合个人开发和快速验证，一条命令拉模型，一条命令启动服务；vLLM 适合生产环境，吞吐高、支持高并发。
以 Ollama 为例，启动和调用非常直接：
# 拉取模型，具体模型名按可用列表选择 ollama pull qwen2.5 # 启动本地 OpenAI 兼容接口 ollama serve默认端口是 11434，Ollama 还提供了 OpenAI 兼容的/v1/chat/completions路由，所以原有代码只需要改base_url就能切到本地模型。这个能力非常实用：
from openai import OpenAI client = OpenAI( base_url="http://127.0.0.1:11434/v1" ) resp = client.chat.completions.create( model="qwen2.5", messages=[{"role": "user", "content": "解释一下本地模型部署的显存估算方法"}], ) print(resp.choices[0].message.content)7.2 显存和性能观察方法
本地部署最需要关注的是显存占用和推理延迟。如果你用 NVIDIA 显卡，可以通过nvidia-smi实时查看显存变化：
watch -n 1 nvidia-smi观察要点：
- 模型加载后显存占用是否稳定，是否存在显存溢出导致进程被杀。
- 并发请求时显存增长多少，超过显存容量会导致 OOM。
- 降低并发、减小上下文长度、使用量化模型都可以降低显存占用。
需要强调的是，不同模型、不同量化方式、不同上下文长度的显存要求差异非常大。比如 7B 模型和 70B 模型完全不是一个量级，选型时必须先根据模型官方文档查阅推荐配置，再结合自己的显卡实测。不要盲目相信一张图上的“4G 显存可跑”。
7.3 什么场景适合本地模型
本地部署的核心优势是数据不出域、调用成本稳定、可深度定制。如果你的业务有下面的特征，更值得考虑本地：
- 涉及用户个人敏感数据、医疗健康信息、企业内部机密。
- 调用量非常稳定且持续，按 token 付费长期不划算。
- 需要自定义某个领域的提示词策略，或微调模型。
- 实时性要求高，网络抖动不可接受。
本地部署的代价也很明确：模型效果上限、硬件采购成本、运维投入、版本更新都要自己负责。这一块要有心理准备。
8. 常见问题与排查方法
从 API 调用到本地部署，我整理了一份高频问题清单，适合直接收藏。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| API 返回 401 | API Key 错误或过期 | 检查环境变量、平台控制台 | 重新生成 Key 并更新环境变量 |
| API 返回 404 | 模型名不存在或账号无权限 | 核对模型名拼写和控制台权限 | 换成有权限的模型名，或开通对应服务 |
| API 返回 429 | 触发平台速率限制或配额不足 | 查看响应头和平台配额 | 降低并发、增加退避、申请更高配额 |
| 请求超时 | 网络不稳定、内容太长、模型推理慢 | 查看超时日志，分段测试 | 调整 timeout 参数，减短输入，开启流式 |
| 上下文长度超限 | 输入加输出超过模型最大 token | 计算输入 token，看报错信息 | 截断输入、用摘要、改大模型版本 |
| 并发超过配额 | 线程并发设置太高 | 查看平台 RPM/TPM 限制 | 控制线程数、加上限流组件 |
| 显存不足 | 模型参数量太大或上下文过长 | 用 nvidia-smi 观察 | 换量化模型、减小 batch、增开虚拟内存 |
| 本地服务拒绝连接 | 端口没起或防火墙拦截 | 检查进程和端口监听 | 重启服务，开放对应端口 |
| 批量任务卡住 | 队列消费逻辑异常 | 打印每个任务状态和耗时 | 增加超时强制结束、失败单测重跑 |
| 成本突然超支 | 没有预算熔断和缓存 | 查看账单明细和 token 数 | 加配额、加缓存、加价格告警 |
排查时记住一个原则：先看日志，再看监控，最后看账单。日志告诉你程序执行到哪一步，监控告诉你资源是否够，账单告诉你钱花在哪了。
9. 最佳实践与使用建议
9.1 多模型冗余是生产底线
不要把线上服务绑死在单一模型平台上。至少在一个兼容层里配置两个可用的模型提供方。主模型挂了自动切备用，即使备用模型效果差一点，也比整个服务不可用强。
9.2 数据脱敏比模型选型更优先
不管用闭源 API 还是开源模型，用户请求内容里的手机号、身份证、地址、财务报表等敏感信息都应该先脱敏。闭源 API 的数据处理政策即使写着“默认不训练”，也不等于“数据永不离开服务器”。本地部署同样要防止模型输出本身泄露训练数据，所以高敏感场景下要做输出过滤。
9.3 预算熔断和告警必须提前配
大模型 API 的费用可以非常快地累积。重点不是“省”，而是“可控”。建议配置每日预算线、单任务最大 token 限制、单用户调用频率限制。一旦触发阈值，立刻切到备用模型或停止自动调用。
9.4 清晰的目录和日志规范
把模型调用、缓存、输入输出、运行日志分目录管理，任务 ID 贯穿全链路。这样排查一个批量任务失败时，能快速定位是输入问题、模型问题还是网络问题。
9.5 合规与授权意识不能缺
如果业务涉及他人的肖像、声音、版权文本、专利内容，无论底层用哪家模型，都必须先确认授权。生成式 AI 的内容也可能存在事实错误和版权风险，发布前要做人工复核。这些不是“可选项”，是生产底线。
10. 总结与下一步
“70% of AI revenue comes from OpenAI and Anthropic”这个判断，对普通用户来说可能只是新闻热点，但对做技术的我们来说，它意味着模型供应链风险、预算控制压力和架构设计选择。最值得做的事是：马上把模型调用从业务代码里抽离出来，引入兼容层，配置至少两个模型提供方，并做好批量任务的并发控制和预算熔断。
最先验证的功能是：用同一段业务代码，分别通过 OpenAI API、Anthropic API 和本地模型跑通一个完整请求。只要这个跑通了，后续切模型、做批量、控成本就是配置项问题。
最容易踩的坑有两个：一是直接把官方 SDK 写进业务代码，后面换模型等于重写；二是不设预算熔断，一个循环调用把月度预算烧光。这两个坑现在避开，后面能省很多返工时间。
下一步可以继续往这几个方向深入：把兼容层部署成独立网关服务，接入统一鉴权和日志；把批量任务改成异步队列并加可视化监控；对比同一任务在闭源 API 和本地模型上的成本、延迟和效果，形成一份自己的选型数据。这几件事做完，你对 AI 收入集中在哪两家这件事，就不再是旁观者，而是能主动控制成本的技术决策者。
