从零构建生产级AI应用开发平台：Agent编排、多供应商接入与RAG扩展实战

1. 为什么我要自己搭一套 AI 应用开发平台

先说结论：市面上能用的 Agent 编排工具我基本都试过一遍，从最早的纯 Prompt 拼接，到后来的工作流引擎，再到最近半年集中爆发的各种 Agent 框架，踩的坑比写的代码还多。最后促使我自己动手做 XXL-AI 的原因很朴素——没有一个平台能同时把「编排灵活度」「多供应商切换」「扩展能力」和「工程化落地」这四件事都做好。

大部分工具要么编排很爽但扩展性差，想接个自定义知识库得改源码；要么扩展点很多但工程化一塌糊涂，本地跑得欢，一上生产就各种超时、状态丢失、日志找不到。更别提多供应商切换了，很多平台绑定单一模型服务，想换个模型得把整个调用链重写一遍。

XXL-AI 这个项目就是冲着这几个痛点去的。它的定位很明确：一个面向开发者的 AI 应用开发平台，核心能力包括 Agent 编排、多供应商接入、MCP + SKILL + RAG 三层扩展体系，以及一套能直接上生产的工程化底座。说白了，它不是给产品经理拖拽玩的低代码玩具，而是给真正要写代码、要部署、要维护的开发者用的。

这篇文章我会把这套平台的完整设计思路、核心模块的实现细节、实操过程中踩过的坑，以及一些只有真正跑过生产环境才知道的经验，全部摊开讲。适合正在选型 Agent 框架的团队、想自己搭一套 AI 中台的工程师，以及被各种「Demo 很美好、上线就崩」折磨过的同行。读完你至少能搞清楚：一个能落地的 AI 应用平台，到底该长什么样。

2. 整体架构设计与核心思路拆解

2.1 为什么是「编排 + 多供应商 + 三层扩展」这个组合

先解释一下这个架构组合背后的逻辑。Agent 编排解决的是「多个 AI 能力怎么协同」的问题；多供应商解决的是「模型从哪来、怎么切换、怎么容灾」的问题；MCP + SKILL + RAG 三层扩展解决的是「平台能力边界怎么突破」的问题；工程化底座解决的是「怎么从 Demo 走到生产」的问题。这四件事缺一不可，少任何一个，平台都会在某个阶段卡住。

我见过太多项目只做了编排，结果模型一换全部重写；也见过只做了 RAG，结果多个 Agent 之间没法协作。XXL-AI 的设计原则是分层解耦、各司其职：编排层只管流程调度，不关心模型是谁；供应商层只管统一接口，不关心上层怎么用；扩展层只管能力注入，不关心调用方是谁。这样任何一层出问题或者要替换，都不会牵连其他层。

具体到技术选型，编排层我用的是基于 DAG（有向无环图）的节点式编排，而不是简单的链式调用。原因很直接：链式调用只能线性执行，一旦遇到「根据中间结果决定下一步走哪个分支」这种需求就废了。DAG 天然支持条件分支、并行执行、循环重试，这才是真实业务场景需要的东西。

2.2 分层架构的职责边界

整个平台我分成了五层，从下往上依次是：

这个分层不是拍脑袋定的，而是根据「变更频率」来划分的。基础设施层最稳定，几个月不动一次；供应商适配层中等，新模型出来就要加适配器；扩展能力层和编排执行层变更最频繁，几乎每周都在迭代。把变更频率相近的东西放在一层，能最大程度减少改动时的连锁反应。

提示：分层的时候一定要想清楚「谁依赖谁」。我的原则是上层依赖下层，下层绝不反向依赖上层。供应商适配层不知道编排层的存在，扩展层也不知道谁在调用它。这样任何一层都能独立测试、独立替换。

2.3 与主流方案的差异化取舍

市面上主流的 Agent 框架我基本都研究过，它们的共同问题是把编排和运行时耦合得太紧。编排定义和执行引擎绑死，导致你想换个执行策略（比如从同步改成异步、从单机改成分布式）就得重写编排逻辑。

XXL-AI 的做法是把编排定义（DSL）和执行引擎彻底分离。编排定义就是一份 JSON 或 YAML，描述节点和连线关系；执行引擎负责解释这份定义并调度执行。这样做的好处是：同一份编排定义，可以用单机引擎跑，也可以用分布式引擎跑；可以同步执行，也可以异步执行；甚至可以拿去做静态分析、可视化渲染。这种解耦带来的灵活性，在实际迭代中价值巨大。

另一个差异点是扩展体系的分层设计。很多平台把 MCP、插件、知识库混在一起，统称「工具」。但它们的本质完全不同：MCP 是标准化的外部服务协议，SKILL 是可复用的能力封装，RAG 是知识检索增强。混在一起会导致调用方式混乱、权限管理困难。XXL-AI 把它们分成三层，每层有独立的注册、发现、调用机制，互不干扰。

3. 核心模块的细节解析与实操要点

3.1 Agent 编排引擎：DAG 节点设计与状态管理

编排引擎是整个平台的心脏，我重点讲几个关键设计。

节点类型的设计。我把节点分成了五类：输入节点、模型节点、工具节点、条件节点、输出节点。输入节点负责接收外部参数；模型节点调用 LLM；工具节点调用 MCP 或 SKILL；条件节点做分支判断；输出节点汇总结果。每类节点有统一的接口定义，包括execute(context)和validate()两个方法。这样新增节点类型只需要实现这两个方法，不用改引擎核心。

上下文传递机制。节点之间怎么传数据是个容易踩坑的地方。我一开始用的是「全局共享上下文」，所有节点读写同一个大对象。结果很快出问题：节点 A 改了某个字段，节点 B 读到的可能是被污染的值，调试起来极其痛苦。后来改成了基于引用的显式传递：每个节点声明自己需要哪些输入（从上游节点的输出里取），产出哪些输出（写到自己的命名空间下）。这样数据流向清晰，也方便做静态校验。

状态持久化。长流程执行到一半挂了怎么办？我的方案是每个节点执行完就把状态快照存到关系库。快照包含当前上下文、已执行节点列表、待执行节点列表。恢复时从快照重建执行状态，跳过已完成的节点。这里有个细节：快照要存「节点执行结果」而不是「节点执行动作」，因为动作可能不幂等，重放会出问题。

# 节点执行的核心逻辑（简化版） def execute_node(node, context): # 1. 从上下文提取输入 inputs = extract_inputs(node.input_refs, context) # 2. 校验输入 node.validate(inputs) # 3. 执行节点逻辑 result = node.execute(inputs) # 4. 写入输出到命名空间 context.set(node.output_namespace, result) # 5. 持久化快照 snapshot = build_snapshot(context, node.id) state_store.save(snapshot) return result

注意：快照持久化会带来性能开销。我的经验是只在关键节点做持久化，比如模型调用、外部工具调用这些耗时且可能失败的操作。纯计算节点可以跳过，失败重跑成本很低。

3.2 多供应商接入：统一接口与智能路由

多供应商这块，核心是抽象出一个足够通用的模型调用接口，让上层完全感知不到底层是哪家模型。

接口设计上，我定义了ChatCompletion、Embedding、Rerank三类能力，每类有统一的入参和出参格式。不同供应商的适配器负责把统一格式转换成各家自己的格式。比如有的供应商用messages数组，有的用prompt字符串，适配器里做转换就行。

智能路由是这块的亮点。路由策略支持四种：

• 优先级路由：按配置的优先级顺序尝试，失败自动降级到下一个
• 权重路由：按权重分配流量，用于灰度或负载均衡
• 成本路由：优先选成本低的供应商，超预算时切换
• 能力路由：根据任务类型选最合适的模型，比如代码任务走代码模型

路由配置我放在数据库里，支持热更新。这样线上调整路由策略不用重启服务。

# 路由配置示例 routes: - name: default-chat strategy: priority providers: - name: provider-a model: model-x priority: 1 timeout: 30s - name: provider-b model: model-y priority: 2 timeout: 30s fallback: true retry: 2

限流和熔断也是必须的。每个供应商独立限流，避免一个供应商被打挂影响全局。熔断用滑动窗口统计失败率，超过阈值自动摘除，半开状态试探恢复。

3.3 MCP 扩展：标准化外部能力接入

MCP 这块我踩的坑最多，重点讲讲。

MCP 的本质是一套标准化的协议，让外部服务能以统一的方式暴露自己的能力。平台作为 MCP 客户端，连接各种 MCP 服务端，把服务端提供的能力注册成平台可调用的工具。

接入流程分三步：发现、注册、调用。发现阶段，客户端连接服务端，拉取能力列表（tools、resources、prompts）；注册阶段，把能力信息存到平台的工具注册表；调用阶段，根据工具名找到对应的 MCP 服务端，发起调用。

这里有个关键设计：MCP 工具的调用要支持流式输出。很多 MCP 服务端返回的是流式数据，如果平台不支持流式，用户体验会很差。我的做法是在工具节点里加一个stream标志，开启后调用结果会以 SSE 形式实时推给前端。

# MCP 工具调用（支持流式） async def call_mcp_tool(server, tool_name, args, stream=False): client = get_mcp_client(server) if stream: async for chunk in client.call_tool_stream(tool_name, args): yield chunk else: result = await client.call_tool(tool_name, args) yield result

提示：MCP 服务端的连接要池化，不要每次调用都新建连接。我一开始没做池化，高并发下连接数暴涨，直接把服务端打挂了。后来加了连接池，限制最大连接数，问题解决。

3.4 SKILL 体系：可复用能力的封装与编排

SKILL 和 MCP 的区别在于：MCP 是外部服务的标准化接入，SKILL 是平台内部能力的封装复用。一个 SKILL 可以是一段 Prompt 模板、一个多步骤的编排片段、或者一段自定义代码。

SKILL 的设计目标是让常用能力可以像积木一样拼装。比如「总结文档」这个 SKILL，内部可能包含「分段」「逐段总结」「合并总结」三个步骤。用户不需要知道内部实现，直接调用这个 SKILL 就行。

SKILL 的定义用 YAML 描述：

name: summarize-document version: 1.0.0 description: 对长文档进行分段总结 inputs: - name: document type: string required: true - name: max_length type: int default: 500 outputs: - name: summary type: string steps: - id: split type: tool tool: text-splitter inputs: text: ${inputs.document} chunk_size: 2000 - id: summarize_each type: model model: default-chat prompt: "总结以下内容：${item}" foreach: ${steps.split.output} - id: merge type: model model: default-chat prompt: "合并以下总结：${steps.summarize_each.output}"

SKILL 支持嵌套调用，一个 SKILL 里可以调用另一个 SKILL。但要注意避免循环依赖，我在注册时做了依赖图检测，有环直接拒绝注册。

3.5 RAG 检索增强：从知识库到上下文注入

RAG 这块我重点讲检索质量和上下文注入策略，因为这两点直接决定 RAG 效果。

检索质量取决于三个环节：切分、向量化、召回。切分策略我试过固定长度、按段落、按语义三种。固定长度最简单但效果最差，容易把一句话切断；按段落好一些但段落长度不均；按语义效果最好但成本高。我的方案是混合策略：先按段落切，段落过长再按语义切，段落过短则合并。

向量化用平台统一的 Embedding 接口，支持多供应商。这里有个坑：不同供应商的向量维度可能不同，不能混用。我在向量库里按供应商+模型维度分 collection，避免维度冲突。

召回策略我用了向量召回 + 关键词召回 + 重排的三段式。向量召回负责语义相似，关键词召回负责精确匹配，重排模型负责精排。实测下来，三段式比单纯向量召回的效果提升明显，尤其是专业术语多的场景。

# RAG 检索流程 def retrieve(query, top_k=5): # 1. 向量召回 vector_results = vector_store.search( embed(query), top_k=top_k * 2 ) # 2. 关键词召回 keyword_results = keyword_index.search(query, top_k=top_k * 2) # 3. 合并去重 merged = merge_and_dedup(vector_results, keyword_results) # 4. 重排 reranked = rerank_model.rerank(query, merged, top_k=top_k) return reranked

上下文注入也有讲究。不是召回的内容全部塞进 Prompt 就好，塞太多会稀释关键信息，还会超 token 限制。我的做法是按相关性排序后动态截断，同时给每段内容加上来源标注，方便模型引用。

4. 完整实操流程与关键环节实现

4.1 环境准备与依赖安装

先把环境跑起来。XXL-AI 的后端是 Python，前端是 React，依赖的中间件包括 PostgreSQL（关系库）、Redis（缓存）、Milvus 或 Qdrant（向量库）。

# 克隆项目 git clone https://github.com/your-org/xxl-ai.git cd xxl-ai # 后端依赖 cd backend python -m venv venv source venv/bin/activate pip install -r requirements.txt # 前端依赖 cd ../frontend npm install # 启动中间件（用 docker-compose） cd .. docker-compose up -d postgres redis milvus

配置文件在backend/config/settings.yaml，重点配这几项：

database: url: postgresql://user:pass@localhost:5432/xxlai redis: url: redis://localhost:6379/0 vector_store: type: milvus host: localhost port: 19530 providers: - name: provider-a api_key: ${PROVIDER_A_KEY} base_url: https://api.provider-a.com

注意：API Key 不要硬编码在配置文件里，用环境变量注入。我见过太多项目把 Key 提交到 Git 仓库，结果被扫出来盗用。

4.2 第一个 Agent 编排：从零到跑通

我拿一个实际场景来演示：用户提问 → 检索知识库 → 判断是否需要调用工具 → 生成回答。

第一步，定义编排 DSL：

name: qa-with-rag nodes: - id: input type: input schema: question: string - id: retrieve type: tool tool: rag-retrieve inputs: query: ${input.question} top_k: 5 - id: decide type: condition expression: ${retrieve.output.score} > 0.7 branches: - when: true next: generate - when: false next: call_tool - id: call_tool type: tool tool: web-search inputs: query: ${input.question} - id: generate type: model model: default-chat prompt: | 基于以下资料回答问题： ${retrieve.output.docs} 问题：${input.question} - id: output type: output inputs: answer: ${generate.output} edges: - from: input to: retrieve - from: retrieve to: decide - from: call_tool to: generate - from: generate to: output

第二步，注册编排并测试：

curl -X POST http://localhost:8000/api/workflows \ -H "Content-Type: application/json" \ -d @qa-with-rag.yaml curl -X POST http://localhost:8000/api/workflows/qa-with-rag/run \ -H "Content-Type: application/json" \ -d '{"question": "XXL-AI 支持哪些扩展方式？"}'

第三步，观察执行日志。平台会记录每个节点的输入、输出、耗时。如果某个节点失败，日志里能看到具体原因。

4.3 接入一个 MCP 服务端

假设我们要接入一个提供「天气查询」能力的 MCP 服务端。

# mcp-servers.yaml servers: - name: weather-server transport: stdio command: python args: ["-m", "weather_mcp_server"] tools: - name: get_weather description: 查询指定城市天气 parameters: city: type: string required: true

平台启动时会自动连接这个服务端，拉取工具列表并注册。之后在编排里就能直接用了：

- id: weather type: tool tool: get_weather inputs: city: ${input.city}

提示：MCP 服务端的启动命令要配超时和重试。有些服务端启动慢，平台如果等不及会报连接失败。我一般配 30 秒超时，失败重试 3 次。

4.4 构建一个 RAG 知识库

RAG 知识库的构建分四步：上传文档 → 切分 → 向量化 → 入库。

from xxlai.rag import KnowledgeBase kb = KnowledgeBase(name="product-docs") # 上传文档 kb.upload("docs/product-manual.pdf") # 配置切分策略 kb.configure_splitter( strategy="hybrid", chunk_size=1000, chunk_overlap=200 ) # 配置向量化模型 kb.configure_embedding(provider="provider-a", model="embedding-v2") # 执行索引构建 kb.build_index()

构建完成后可以测试检索：

results = kb.search("如何配置多供应商路由", top_k=3) for r in results: print(f"score: {r.score}, content: {r.content[:100]}")

实测下来，切分策略对检索质量影响最大。我建议先用混合策略跑一遍，看看召回结果，再根据实际情况调整 chunk_size 和 overlap。overlap 一般设 chunk_size 的 10% 到 20%。

5. 常见问题与排查技巧实录

5.1 编排执行类问题

问题一：节点执行超时，但日志里看不到具体卡在哪。

这个我遇到过好几次。原因是节点执行没有加超时控制，某个外部调用卡死了，整个流程就挂在那里。解决方案是给每个节点加独立的超时配置，超时后抛出异常，由引擎决定是重试还是走 fallback 分支。

- id: slow_tool type: tool tool: external-api timeout: 10s on_timeout: fallback

问题二：条件节点判断结果不符合预期。

八成是表达式解析的问题。我的条件节点用的是受限的表达式引擎，只支持基本的比较和逻辑运算。如果表达式里引用了不存在的字段，会返回 None，导致判断异常。排查方法是在条件节点里打印表达式的求值过程，看看每个变量实际是什么值。

问题三：并行节点执行结果顺序错乱。

DAG 里并行执行的节点，完成顺序是不确定的。如果下游节点依赖多个并行节点的输出，要确保所有上游都完成后再执行。我的引擎用依赖计数来管理：每个节点维护一个未完成依赖数，归零时才触发执行。

5.2 供应商调用类问题

问题一：切换供应商后，同样的 Prompt 输出质量差异很大。

这是正常的，不同模型的「脾气」不一样。解决方案是为每个供应商维护独立的 Prompt 模板，或者用更通用的 Prompt 写法。我一般会在 Prompt 里明确输出格式要求，减少模型自由发挥的空间。

问题二：某个供应商频繁超时，但其他供应商正常。

先排查是不是网络问题，再看是不是该供应商限流了。平台的限流是按供应商独立配置的，如果配置的 QPS 太低，正常请求也会被限。另外要注意供应商的并发限制，有些供应商限制同时进行的请求数，超了会直接拒绝。

问题三：流式输出中断，前端收到一半就没了。

流式输出对网络稳定性要求高，中间任何一环断了都会中断。我的做法是在流式输出的同时做缓冲，如果连接断了，前端可以从断点续传。另外要设置合理的超时，流式输出不能按普通请求的超时来算。

5.3 RAG 检索类问题

问题一：检索结果相关性差，答非所问。

先检查切分是否合理。如果 chunk 太大，一个 chunk 里混了多个主题，检索时容易匹配到不相关的内容。如果 chunk 太小，信息不完整，模型也没法回答。我一般从 chunk_size=1000 开始调，根据文档类型微调。

问题二：知识库更新后，检索还是返回旧内容。

这是缓存问题。向量库和关键词索引都有缓存，更新文档后要主动刷新缓存。我的做法是文档更新时触发索引重建，重建完成后清空相关缓存。

问题三：多语言文档检索效果差。

不同语言的向量空间不一样，混在一起检索效果会打折。解决方案是按语言分 collection，检索时根据查询语言路由到对应的 collection。

5.4 常见问题速查表

提示：排查问题的第一原则是先看日志，再看监控，最后才猜。我见过太多人上来就改代码，结果改了半天发现是配置问题。平台的日志要打全，关键节点的输入输出、耗时、异常堆栈都要记录。

6. 工程化底座的落地经验

6.1 可观测性：日志、指标、链路追踪

工程化底座里，可观测性是最容易被忽视但最重要的部分。没有可观测性，线上出问题就是盲人摸象。

我的方案是三件套：结构化日志 + 指标监控 + 链路追踪。日志用 JSON 格式，每条日志带 trace_id、node_id、耗时等字段；指标用 Prometheus 采集，重点监控 QPS、延迟、错误率、token 消耗；链路追踪用 OpenTelemetry，把一次请求经过的所有节点串起来。

# 结构化日志示例 logger.info("node_executed", extra={ "trace_id": ctx.trace_id, "node_id": node.id, "duration_ms": duration, "status": "success", "tokens": result.tokens })

这样出问题时，先通过 trace_id 找到完整链路，再定位到具体节点，最后看该节点的详细日志。整个排查过程从原来的半小时缩短到几分钟。

6.2 配置管理：环境隔离与热更新

配置管理我踩过一个坑：开发环境的配置不小心带到生产环境，导致生产连了测试数据库。后来我做了严格的环境隔离：每个环境独立的配置文件，通过环境变量ENV决定加载哪个。生产环境的配置只能通过配置中心下发，不允许本地修改。

热更新也很重要。路由策略、限流阈值、Prompt 模板这些经常要调整，如果每次都要重启服务，运维成本太高。我的做法是配置存数据库，监听变更事件，变更后自动刷新内存中的配置。

6.3 部署与扩缩容

部署我推荐容器化 + K8s。每个组件独立打包成镜像，通过 K8s 编排。编排引擎是无状态的，可以水平扩容；向量库和关系库是有状态的，用 StatefulSet 部署。

扩缩容策略上，编排引擎按 CPU 和队列长度扩容，队列积压超过阈值就加副本；供应商适配层按 QPS 扩容，QPS 上来了就加实例。缩容要保守一些，避免频繁抖动。

注意：扩容时要注意冷启动问题。新实例启动后需要预热，比如加载模型、建立连接池。如果不预热直接接流量，前几个请求会很慢。我的做法是启动后先跑一遍健康检查，通过后再加入负载均衡。

7. 一些只有踩过坑才知道的经验

先说一个最容易被忽视的点：Agent 编排的调试体验比功能本身更重要。我一开始只关注功能实现，结果调试一个复杂编排要花半天。后来加了单步执行、断点、变量快照这些调试功能，效率提升了好几倍。如果你也在做类似平台，一定要把调试体验做好。

另一个经验是不要过度设计。我一开始想做一个「万能」的编排引擎，支持各种花哨的功能，结果复杂度爆炸，维护成本极高。后来砍掉了很多不常用的功能，聚焦核心场景，反而更稳定。平台的价值在于解决 80% 的常见问题，而不是覆盖 100% 的边缘场景。

关于 RAG，我的体会是检索质量比模型能力更重要。同样的模型，检索做得好，回答质量天差地别。与其花时间调 Prompt，不如先把切分和召回做好。

最后分享一个小技巧：给每个 Agent 编排加一个「干跑」模式。干跑模式下不实际调用模型和工具，只走流程逻辑，返回模拟结果。这样测试编排逻辑时不用消耗 token，速度也快很多。这个功能在开发阶段帮我省了大量成本。