1. 从一个真实的技术选型困境说起做 AI Data Agent 这个方向两年多我前后主导过三套不同形态的 Agent 系统落地一套是早期基于开源 Agent Framework 快速搭起来的原型一套是后来为了接入公司内部十几个数据源而自研的通用 Runtime Plugin 架构还有一套是给业务方做 POC 时临时拼的轻量方案。这三套东西踩过的坑、走过的弯路基本覆盖了今天大家在 Agent 技术选型上会遇到的绝大多数纠结。最近半年被问得最多的问题就是做 AI Data Agent到底该选通用 Agent Runtime Plugin 的模式还是直接用现成的 Agent Framework这个问题看起来是个技术选型问题实际上背后牵扯的是你对业务边界的判断、对团队工程能力的评估、以及对未来扩展性的预期。选错了轻则重构一遍重则整个项目方向跑偏浪费几个月的人力。我先把结论摆出来免得有人看到一半就急着下判断没有绝对正确的答案只有匹配当前阶段的选择。但选择背后有一套可以量化的判断逻辑这套逻辑才是真正值钱的东西。这篇文章我会把两套方案的核心差异、适用场景、实操细节、踩坑经验全部拆开讲清楚尤其是 AI Data Agent 这个特定场景下的取舍因为数据类 Agent 和通用对话 Agent 在架构诉求上差别非常大。先明确一下讨论范围。这里说的Agent Runtime指的是负责 Agent 生命周期管理、任务调度、状态维护、工具调用编排的运行时底座Plugin指的是挂载在 Runtime 上的具体能力单元比如查数据库、调 API、跑 SQL、做数据清洗这些。而Agent Framework指的是像 LangChain、LlamaIndex、AutoGen 这类提供了完整抽象层、开箱即用的开发框架。两者的核心区别在于Runtime Plugin 是底座 插槽的思路Framework 是全家桶的思路。适合读这篇的人正在做 AI Data Agent 技术选型的架构师、需要给团队定技术路线的技术负责人、以及想搞清楚这两套方案到底差在哪的一线开发。不管你是刚起步做 POC还是已经在生产环境跑了一段时间想优化下面的内容应该都能对上你的场景。2. 两套方案的本质差异到底在哪2.1 控制权归属谁掌握核心逻辑这是两套方案最根本的分歧点也是后面所有差异的源头。用 Agent Framework 的时候核心编排逻辑是框架给你的。框架定义了 Agent 怎么思考、怎么调用工具、怎么维护记忆、怎么做多轮对话。你做的事情是在框架给定的抽象里填内容——写 Prompt、注册 Tool、配置 Memory。好处是上手快坏处是当你的业务逻辑和框架的假设不一致时你要么绕要么改框架源码要么换方案。通用 Runtime Plugin 的模式反过来核心编排逻辑是你自己写的。Runtime 只负责最底层的调度、状态、生命周期Plugin 只负责单一能力中间的编排逻辑、决策逻辑、数据流转逻辑全部由你掌控。这意味着你可以为 AI Data Agent 定制非常贴合业务的控制流比如先做数据探查再决定用哪个查询策略查询失败后走降级路径这种复杂逻辑写起来非常自然。我举个具体例子。做数据 Agent 时经常遇到一个场景用户问上个月华东区的销售趋势怎么样Agent 需要先判断这句话涉及哪些表、哪些字段然后生成 SQL执行如果结果为空还要判断是数据问题还是查询问题最后做可视化。用 Framework 的话这套流程你得拆成多个 Chain 或者 Agent 节点节点间的状态传递依赖框架的机制调试起来很痛苦。用 Runtime Plugin 的话你可以把意图解析Schema 匹配SQL 生成执行结果校验做成五个 PluginRuntime 负责按你的规则串起来每一步的输入输出都是你自己定义的结构出问题一眼就能定位。2.2 扩展成本加一个新能力要动多少东西AI Data Agent 有个特点数据源会不断增加。今天接 MySQL明天接 ClickHouse后天要接对象存储里的 Parquet 文件再往后可能还要接实时流。扩展成本直接决定了你的迭代速度。Framework 模式下加一个新数据源通常要写一个新的 Tool然后注册到 Agent 里还要考虑这个 Tool 的描述怎么写才能让模型正确调用。问题在于当 Tool 数量超过十几个之后模型的工具选择准确率会明显下降而且 Framework 的工具注册机制往往不支持复杂的路由逻辑你很难做根据数据源类型动态加载工具集这种优化。Runtime Plugin 模式下Plugin 是独立注册、独立加载的。你可以做插件市场式的管理按业务域、按数据源类型分组Runtime 根据当前任务上下文只加载相关的 Plugin 子集。这个差异在数据源超过 20 个之后会变得非常明显——我实测过同样的模型Plugin 分组加载比全量 Tool 注册的工具选择准确率能高出 30% 以上。2.3 调试与可观测性出问题能不能快速定位数据 Agent 的调试难度远高于普通对话 Agent因为它的执行链路长、涉及外部系统多、失败原因复杂。一条查询失败可能是意图理解错了、可能是 Schema 匹配错了、可能是 SQL 语法问题、可能是权限问题、可能是数据源本身挂了。Framework 的调试体验普遍一般。因为框架做了太多封装你看到的是一层抽象后的日志真正的执行细节被藏在框架内部。想加详细的埋点往往要侵入框架代码。而且 Framework 的调用链是框架自己控制的你很难在中间插入自定义的监控逻辑。Runtime Plugin 因为控制权在你手里可观测性可以做得很细。每个 Plugin 的输入输出、耗时、异常都能独立记录Runtime 层面可以做全链路追踪哪个环节慢、哪个环节错一目了然。我们生产环境的 Data Agent 就是靠这套机制把平均故障定位时间从小时级压到了分钟级。2.4 方案对比速查表维度Agent Framework通用 Runtime Plugin上手速度快几天出原型慢需要先搭底座控制粒度粗受框架抽象限制细完全自主扩展成本随 Tool 增多而上升插件化扩展平滑调试难度中高封装层厚低链路透明定制能力弱需绕框架强任意定制团队要求低会调 API 即可高需要架构能力适合阶段POC、快速验证生产、规模化长期维护受框架版本影响自主可控这张表不是让你直接照着选而是让你看清楚每个维度的代价。下面我会结合 AI Data Agent 的具体场景讲怎么把这些维度转化成实际的决策依据。3. AI Data Agent 场景下的选型判断逻辑3.1 先问自己三个问题在动手选型之前我建议先老老实实回答三个问题答案基本能决定你该走哪条路。第一个问题你的数据源数量和类型未来半年会增长到什么规模如果只是接一两个固定的数据库Framework 完全够用别过度设计。但如果预期会接五种以上异构数据源而且还在持续增加那 Runtime Plugin 的扩展优势会很快体现出来。我见过太多团队一开始用 Framework 图快结果数据源加到第八个的时候被迫重构前面省的时间全吐出来了。第二个问题你的团队有没有能力维护一套 Runtime这是最现实的问题。Runtime Plugin 模式对团队的工程能力要求明显更高你需要有人懂任务调度、懂状态管理、懂插件隔离、懂可观测性建设。如果团队里全是算法背景、工程能力偏弱硬上 Runtime 会非常痛苦。反过来如果团队有扎实的后端工程底子Runtime 这套东西对他们来说并不复杂。第三个问题这个 Agent 是 demo 还是要上生产demo 阶段追求的是快速验证想法Framework 的开箱即用价值巨大。但一旦要上生产面对并发、稳定性、可维护性的要求Framework 的封装反而会变成负担。我的经验是demo 用 Framework生产用 Runtime中间做一次迁移这个路径对大多数团队都成立。3.2 数据 Agent 的特殊性为什么它比通用 Agent 更挑架构通用对话 Agent 的核心是理解 生成链路相对短。数据 Agent 的核心是理解 规划 执行 校验 呈现链路长得多而且中间每一步都可能失败。更关键的是数据 Agent 对准确性的要求远高于通用 Agent。聊天机器人说错一句话无所谓数据 Agent 查错一个数可能导致业务决策失误。这就要求架构上必须支持严格的校验、重试、降级机制而这些机制在 Framework 里往往很难优雅地实现。还有一个容易被忽略的点数据 Agent 的上下文管理更复杂。它需要维护的不只是对话历史还有当前查询涉及的表结构、字段含义、历史查询模式、用户偏好等。这些上下文怎么组织、怎么按需加载、怎么避免污染Framework 的通用 Memory 机制往往不够用需要你自己设计。3.3 一个可量化的决策模型我把上面的判断逻辑整理成一个简单的打分模型你可以对着自己的情况算一下。每个维度 1-5 分分数越高越倾向于 Runtime Plugin。评估维度1分倾向Framework5分倾向Runtime你的得分数据源数量预期1-2个10个以上团队工程能力算法为主工程扎实项目阶段POC验证生产规模化定制化需求标准流程高度定制可观测性要求一般严格长期维护预期短期长期演进总分低于 15 分Framework 是更理性的选择15-22 分可以考虑混合方案先用 Framework 验证再用 Runtime 重构高于 22 分直接上 Runtime Plugin别犹豫。这个模型不是绝对标准但它能帮你把模糊的直觉变成相对清晰的判断。我自己的项目当时算下来是 26 分所以走了 Runtime 路线事后看这个决策是对的。4. Runtime Plugin 架构的实操落地4.1 Runtime 层要解决的核心问题如果你决定走 Runtime Plugin 路线Runtime 层需要解决四个核心问题这四个问题解决好了整个架构就立住了。任务调度与生命周期管理。Agent 的一次任务执行从接收请求到返回结果中间会经历多个阶段。Runtime 要负责创建任务上下文、按编排逻辑调度 Plugin、处理 Plugin 的返回、管理任务状态、处理超时和异常。这块我建议用状态机来建模每个任务是一个状态机实例状态流转清晰可控。我们用的是自己实现的一个轻量状态机比引入工作流引擎更灵活。上下文与状态管理。任务执行过程中产生的中间数据、Plugin 的输入输出、临时的推理结果都需要有个地方存。我的做法是给每个任务分配一个独立的 Context 对象所有 Plugin 通过 Context 读写数据Runtime 负责 Context 的生命周期。这样 Plugin 之间解耦数据流转清晰。注意 Context 要做大小控制避免把大结果集全塞进去导致内存爆炸。Plugin 注册与发现。Plugin 要能动态注册、按需加载。我们设计了一个 Plugin Registry每个 Plugin 声明自己的元信息名称、能力描述、输入输出 Schema、依赖关系Runtime 根据任务需求从 Registry 里挑选合适的 Plugin 加载。这样新增数据源只需要写一个新 Plugin 注册进去不用动 Runtime 代码。可观测性基础设施。这块前面提过是 Runtime 相对 Framework 的最大优势之一。每个 Plugin 的执行都要埋点记录输入、输出、耗时、异常。Runtime 层面做全链路追踪给每个任务分配 Trace ID所有日志和指标都带上这个 ID。生产环境排查问题时一个 Trace ID 就能还原整个执行链路。4.2 Plugin 的设计原则Plugin 设计得好不好直接决定了整个系统的可维护性。我总结了四条原则都是踩坑踩出来的。单一职责。一个 Plugin 只做一件事。查 MySQL 的 Plugin 就只管查 MySQL不要在里面顺便做数据清洗。清洗是另一个 Plugin 的事。这样 Plugin 可以自由组合复用性高。我早期犯过这个错把查询和转换揉在一个 Plugin 里后来想换个清洗逻辑发现要改查询 Plugin非常别扭。接口标准化。所有 Plugin 遵循统一的接口约定输入是一个标准化的请求对象输出是一个标准化的响应对象。这样 Runtime 调度 Plugin 的时候不需要关心具体实现。我们的约定是输入包含params业务参数和context任务上下文引用输出包含status、data、error、metadata四个字段。无状态优先。Plugin 尽量设计成无状态的所有状态通过 Context 传递。这样 Plugin 可以并发执行也方便测试。确实需要状态的场景比如连接池把状态封装在 Plugin 内部对外仍然表现为无状态。失败可预期。Plugin 要明确声明自己可能抛出的异常类型Runtime 根据异常类型决定重试、降级还是终止。不要用笼统的 Exception 糊弄异常分类越细Runtime 的处理策略就能越精准。4.3 一个数据查询 Plugin 的完整实现示例光讲原则太虚我拿一个实际的 MySQL 查询 Plugin 举例把关键代码和设计思路都摊开。from typing import Any, Dict from dataclasses import dataclass dataclass class PluginRequest: params: Dict[str, Any] context: Any dataclass class PluginResponse: status: str # success / failed / retry data: Any error: str metadata: Dict None class MySQLQueryPlugin: name mysql_query description 执行 MySQL 查询输入 SQL 语句返回结果集 input_schema {sql: str, database: str, timeout: int} output_schema {rows: list, columns: list, row_count: int} def __init__(self, connection_pool): self.pool connection_pool def execute(self, request: PluginRequest) - PluginResponse: sql request.params.get(sql) database request.params.get(database) timeout request.params.get(timeout, 30) # 参数校验 if not sql or not database: return PluginResponse( statusfailed, dataNone, errormissing required params: sql or database ) # SQL 安全校验防止危险操作 if self._is_dangerous_sql(sql): return PluginResponse( statusfailed, dataNone, errordangerous sql detected ) try: conn self.pool.get_connection(database) cursor conn.cursor() cursor.execute(fSET SESSION MAX_EXECUTION_TIME{timeout*1000}) cursor.execute(sql) rows cursor.fetchall() columns [desc[0] for desc in cursor.description] return PluginResponse( statussuccess, data{rows: rows, columns: columns, row_count: len(rows)}, metadata{database: database, sql: sql} ) except TimeoutError as e: return PluginResponse(statusretry, dataNone, errorstr(e)) except Exception as e: return PluginResponse(statusfailed, dataNone, errorstr(e)) def _is_dangerous_sql(self, sql: str) - bool: dangerous [drop, truncate, delete, update, insert] sql_lower sql.lower().strip() return any(sql_lower.startswith(kw) for kw in dangerous)这段代码有几个设计点值得说明。第一status有三种值retry表示可重试的失败比如超时Runtime 看到这个状态会自动重试failed表示不可重试的失败直接终止。第二SQL 安全校验放在 Plugin 内部因为这是数据查询特有的约束不该让 Runtime 关心。第三超时通过MAX_EXECUTION_TIME在数据库层面控制比在应用层做超时更可靠。4.4 Runtime 调度逻辑的实现要点Runtime 的调度逻辑是整个系统的中枢我把它简化成一个可读的伪代码重点看编排思路。class AgentRuntime: def __init__(self, plugin_registry, context_store): self.registry plugin_registry self.context_store context_store def run_task(self, task_request): context self.context_store.create(task_request) trace_id context.trace_id try: # 阶段一意图解析 intent self._execute_plugin(intent_parser, {query: task_request.query}, context) if intent.status ! success: return self._handle_failure(intent, context) # 阶段二Schema 匹配 schema self._execute_plugin(schema_matcher, {intent: intent.data}, context) # 阶段三SQL 生成 sql self._execute_plugin(sql_generator, {intent: intent.data, schema: schema.data}, context) # 阶段四执行查询带重试 result self._execute_with_retry(mysql_query, {sql: sql.data, database: schema.data[db]}, context, max_retry3) # 阶段五结果校验 validated self._execute_plugin(result_validator, {result: result.data, intent: intent.data}, context) return self._build_response(validated, context) finally: self.context_store.cleanup(context) def _execute_with_retry(self, plugin_name, params, context, max_retry): for attempt in range(max_retry): resp self._execute_plugin(plugin_name, params, context) if resp.status success: return resp if resp.status failed: return resp # status retry继续循环 context.log(fretry {plugin_name}, attempt {attempt1}) return PluginResponse(statusfailed, dataNone, errormax retry exceeded)这段代码体现了 Runtime 的核心价值编排逻辑完全可控。你可以清楚地看到每个阶段做什么、失败怎么处理、重试策略是什么。换成 Framework这套逻辑要么被拆散到多个 Chain 里要么得用框架的 Agent 机制硬凑可读性和可控性都差很多。5. 常见问题与排查技巧实录5.1 工具选择准确率下降怎么办这是数据 Agent 最常遇到的问题。当 Plugin 数量增多后模型在该调用哪个 Plugin这件事上开始犯错。我遇到过最离谱的一次模型把查订单表的请求路由到了查用户表的 Plugin 上。排查思路是这样的先确认是不是 Plugin 描述写得不够清晰。Plugin 的description是模型做选择的唯一依据描述模糊必然导致误选。我后来强制要求每个 Plugin 的描述必须包含做什么、输入什么、输出什么、什么场景用四个要素准确率立刻上来了。如果描述已经很清楚还是误选那就是 Plugin 数量太多导致的。解决方案是分组加载。给 Plugin 打上业务域标签Runtime 根据当前任务的上下文先判断业务域只加载该业务域下的 Plugin。我们生产环境把 40 多个 Plugin 分成 6 个业务域每次只加载 5-8 个工具选择准确率从 70% 左右提升到了 95% 以上。还有一个技巧是给 Plugin 加负样本描述。比如查订单表的 Plugin 描述里明确写不适用于查询用户信息这种负向提示对模型很有帮助。5.2 长链路执行中途失败怎么恢复数据 Agent 的执行链路动辄七八步中途失败是常态。如果每次都从头重来效率极低。我的做法是在 Context 里做检查点每个 Plugin 执行成功后把结果和状态存下来失败恢复时从最后一个成功的检查点继续。这里有个坑要注意不是所有 Plugin 都适合做检查点。有副作用的 Plugin比如写数据、发通知如果重试会导致重复执行这类 Plugin 要么设计成幂等的要么在检查点机制里特殊标记恢复时跳过。5.3 并发场景下的资源竞争数据 Agent 上生产后必然面临并发。我踩过的最大的坑是数据库连接池被打满。原因是每个查询 Plugin 都去申请连接高并发下连接数瞬间飙升。解决方案有两层。第一层是连接池本身要设置合理的上限和等待超时避免无限等待。第二层是在 Runtime 层面做并发限流根据下游数据源的承载能力限制同时执行的查询数量。我们给每个数据源配置了独立的并发配额超过配额的任务排队等待这样既保护了数据源又保证了整体稳定性。5.4 常见问题速查表问题现象可能原因排查方向解决手段工具选择错误描述不清/数量过多检查 Plugin description优化描述分组加载执行超时下游慢/无超时控制看各环节耗时埋点加超时限流结果不准Schema 匹配错检查 Schema 元数据完善元数据校验内存溢出Context 过大看 Context 大小大结果外存引用重复执行重试无幂等检查 Plugin 副作用幂等设计检查点链路断点异常未捕获看异常日志完善异常分类处理5.5 几个只有踩过才知道的坑第一个坑别在 Plugin 里做重试。重试逻辑应该统一放在 Runtime 层Plugin 只管执行和报告状态。我早期在 Plugin 里写了重试结果 Runtime 又重试一次变成双重