qKnow 3.0:插件化中台如何让企业智能体从"拼凑"走向"组装"

2026-05-28 19 预计阅读时间:1 分钟
来源:oschina.net AI 摘要 原文链接

免责声明:本文为 AI 摘要整理,建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件,不作为官方说明。

预计阅读时间:12 分钟

企业做 AI 应用,最怕的不是模型不够强,而是平台撑不住。架构臃肿耦合、配置流程繁琐、AI 能力碎片散落、数据运营体系缺失——这些老问题让不少团队在智能体落地时反复踩坑。qKnow 3.0 把整个中台架构推倒重来,核心思路只有一个:插件化。能力按需挂载,业务按场景组装,部署按规模弹性伸缩。

老架构的病根:耦合与碎片

传统智能体平台通常把知识检索、对话引擎、任务编排、数据统计全部塞进一个单体服务。表面上看"功能齐全",实际操作时问题频出:

  • 改一个能力要动整个平台。想升级检索算法,得重新部署整套服务,连带对话引擎也跟着重启。
  • 配置流程冗长。新场景上线要改十几个配置文件,环境变量散落各处,排查问题靠 grep。
  • AI 能力零散。不同团队各自接入模型,SDK 版本不一致,Prompt 管理没有统一入口,能力复用率极低。
  • 数据运营断层。智能体上线后,调用量、准确率、用户反馈没有闭环,优化全靠拍脑袋。

qKnow 3.0 的插件化中台,正是针对这些病根做的结构性手术。

插件化中台的核心设计

v3.0 把原来单体拆成三层:基础运行层 → 插件注册层 → 场景编排层。每层职责清晰,互不干扰。

基础运行层

负责进程管理、日志采集、权限鉴权、配置分发。这一层不承载任何业务逻辑,只保证插件能跑起来、能被发现、能被调度。

插件注册层

每个 AI 能力——知识检索、向量计算、对话生成、外部 API 调用、数据报表——都是一个独立插件。插件通过标准接口注册到中台,声明自己的输入输出 schema、依赖的资源、运行模式(同步/异步/流式)。

场景编排层

业务方在编排层把多个插件串成一条流水线,定义触发条件、异常处理、结果聚合。编排配置是声明式的 YAML,不需要写代码。

这种分层的好处很直接:换一个检索插件,对话插件不受影响;加一个报表插件,不用改任何已有编排

实操:用 YAML 编排一个企业知识问答智能体

下面是一个基于 qKnow 3.0 插件化架构的编排配置示例。假设我们要构建一个"产品文档问答"智能体,它需要:检索企业知识库 → 调用大模型生成回答 → 记录对话日志 → 输出统计报表。

注:以下 YAML 结构参考 qKnow 3.0 公开的设计理念编写,具体字段名以官方文档为准。使用前请对照最新版 API 规范调整。

# qKnow 3.0 场景编排配置 — 产品文档问答智能体
agent:
  name: product-doc-qa
  version: 1.0.0
  description: 基于企业知识库的产品文档问答

  # 触发方式:API 调用或定时任务
  triggers:
    - type: api
      path: /api/v1/agent/product-doc-qa/invoke
      method: POST
      input_schema:
        type: object
        properties:
          question:
            type: string
            description: 用户提问内容
          context_filter:
            type: string
            description: 可选,限定检索范围(如产品线名称)
        required: [question]

  # 插件流水线编排
  pipeline:
    steps:
      # Step 1: 知识检索插件
      - plugin: knowledge-retriever
        version: "^2.1"
        config:
          index_name: product-docs-v2
          top_k: 5
          score_threshold: 0.72
          rerank: true
        input_mapping:
          query: "{{ triggers.input.question }}"
          filter: "{{ triggers.input.context_filter }}"
        output_key: retrieved_chunks

      # Step 2: 大模型对话生成插件
      - plugin: llm-generator
        version: "^1.3"
        config:
          model: qwen-plus
          temperature: 0.3
          max_tokens: 2048
          system_prompt: |
            你是产品文档助手。根据检索到的文档片段回答用户问题。
            如果文档中没有相关信息,明确告知用户并建议联系技术支持。
            回答需引用具体文档来源。
        input_mapping:
          prompt: "{{ triggers.input.question }}"
          context: "{{ steps.retrieved_chunks.documents }}"
        output_key: generated_answer

      # Step 3: 对话日志插件(异步写入,不阻塞主流程)
      - plugin: conversation-logger
        version: "^1.0"
        config:
          store: elasticsearch
          index: agent-logs-product-doc-qa
        input_mapping:
          question: "{{ triggers.input.question }}"
          answer: "{{ steps.generated_answer.text }}"
          sources: "{{ steps.retrieved_chunks.document_ids }}"
          latency_ms: "{{ pipeline.total_latency }}"
        output_key: log_status
        async: true

      # Step 4: 统计报表插件(可选,按需挂载)
      - plugin: metrics-reporter
        version: "^1.0"
        config:
          dashboard: product-doc-qa-stats
          metrics: [invoke_count, avg_latency, retrieval_hit_rate]
        input_mapping:
          session_id: "{{ triggers.input.session_id }}"
        output_key: metrics_snapshot
        async: true

  # 异常处理策略
  error_handling:
    default: retry_with_fallback
    rules:
      - plugin: knowledge-retriever
        error_type: index_not_found
        action: fallback_to_web_search
      - plugin: llm-generator
        error_type: rate_limit
        action: queue_and_retry
        max_retries: 3
        retry_delay_ms: 2000

  # 弹性伸缩配置
  scaling:
    min_instances: 2
    max_instances: 10
    scale_up_threshold: 50   # 每实例并发请求超过 50 时扩容
    scale_down_after_idle: 300s

把这份 YAML 放进 qKnow 3.0 的编排管理界面,平台会自动解析插件依赖、校验 schema、拉取对应版本插件并部署流水线。新增一个插件只需加一个 step,不需要改其他步骤的配置

插件开发:最小模板

如果团队需要自定义能力(比如调用内部 ERP 接口),可以按插件标准接口开发。下面是一个 Python 插件的最小模板:

# qKnow 3.0 自定义插件最小模板 — ERP 查询插件
from qknow_plugin import PluginBase, PluginMeta, InputSchema, OutputSchema

class ErpQueryPlugin(PluginBase):
    """查询内部 ERP 系统的订单状态"""

    # 插件元信息,注册到中台时使用
    meta = PluginMeta(
        name="erp-order-query",
        version="1.0.0",
        description="查询 ERP 订单状态,返回订单详情",
        tags=["erp", "order", "internal-api"],
    )

    # 声明输入输出 schema,编排层据此做类型校验和映射
    input_schema = InputSchema({
        "order_id": {"type": "string", "required": True, "description": "ERP 订单号"},
        "detail_level": {"type": "string", "enum": ["summary", "full"], "default": "summary"},
    })

    output_schema = OutputSchema({
        "order_status": {"type": "string"},
        "delivery_date": {"type": "string", "format": "date"},
        "items": {"type": "array", "items": {"type": "object"}},
    })

    def execute(self, input_data: dict) -> dict:
        """核心执行逻辑——实际调用 ERP API"""
        order_id = input_data["order_id"]
        detail_level = input_data.get("detail_level", "summary")

        # 这里替换为真实的 ERP API 调用
        # 示例用模拟数据演示
        erp_response = self._call_erp_api(order_id, detail_level)

        return {
            "order_status": erp_response["status"],
            "delivery_date": erp_response["expected_delivery"],
            "items": erp_response.get("line_items", []),
        }

    def _call_erp_api(self, order_id: str, detail_level: str) -> dict:
        """封装 ERP API 调用,处理认证和异常"""
        import requests

        ERP_BASE_URL = self.config.get("erp_base_url", "https://erp.internal.corp/api")
        API_TOKEN = self.config.get("erp_api_token")

        headers = {"Authorization": f"Bearer {API_TOKEN}"}
        params = {"order_id": order_id, "detail": detail_level}

        try:
            resp = requests.get(
                f"{ERP_BASE_URL}/orders/query",
                headers=headers,
                params=params,
                timeout=10,
            )
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            # 插件标准异常格式,编排层会按 error_handling 规则处理
            raise self.PluginExecutionError(
                error_type="erp_api_error",
                message=f"ERP 查询失败: {e}",
                retryable=True,
            )


# 注册插件(qKnow 加载时自动调用)
if __name__ == "__plugin_main__":
    ErpQueryPlugin.register()

开发完成后,打包为 Docker 镜像推到插件仓库,编排层就能直接引用 erp-order-query 插件。整个流程不需要改中台代码,也不需要重启已有智能体

从拼凑到组装:落地时的几个判断

插件化架构解决了结构性问题,但落地效果取决于怎么用。几个实操判断值得提前想清楚:

哪些能力值得做插件,哪些不该做? 通用性强、变更频率高的能力(检索、生成、日志)适合插件化;高度定制、只在一个场景用的逻辑,硬塞进插件反而增加维护成本,直接写在编排层的自定义脚本里更务实。

插件版本怎么管? 编排 YAML 里用语义版本范围(如 ^2.1)而非锁定版本,保证兼容更新自动生效。但关键生产环境建议锁定具体版本,避免插件静默升级引入行为变化。

异步插件别滥用。 日志、统计这类"写完就行"的能力适合异步,不阻塞主流程。但任何影响最终输出质量的能力(比如二次校验、合规审查)必须同步执行,否则用户拿到未校验的结果。

数据运营要一开始就挂上。 很多团队先做功能再补统计,结果前几个月的调用数据全丢。建议在编排配置里从第一个版本就挂 metrics-reporterconversation-logger,哪怕初期只写基础指标,后续优化才有数据支撑。

弹性伸缩的阈值要按实际压测调。 默认配置里的 scale_up_threshold: 50 是参考值。检索密集型智能体可能 20 并发就该扩容,纯对话型可能 100 并发还撑得住。上线前用真实负载跑一轮压测,再定阈值。


qKnow 3.0 的插件化中台不是"更多功能"的堆叠,而是让每个能力独立演进、按需组合。对企业来说,最实际的收益是:新场景上线从"改平台代码"变成"写一份编排 YAML",交付周期从周级缩到天级。当然,插件化本身也有代价——接口规范要持续维护,插件间依赖要显式声明,版本兼容要持续验证。这些治理成本不会消失,只是从"改代码的隐性成本"变成了"管配置的显性成本",后者至少可控、可追溯。


相关推荐