如果要自己创建一个组件并且使用

要创建一个新的组件类 + 自定义新的组件参数

① agent/component/xxxx.py 在里面继承 ComponentParamBase + ComponentBase 类 这样就可以重用其中的大部分方法和逻辑
② 新组件类需要实现 _run 方法，这个方法会定义组件的具体业务逻辑

BaiduFanyiParam →→ ComponentBase → base.py
★★★ base.py【里面的output是输出为元组(DataFrame) 
再通过序列化转换为(字典)(def as_dict(self))JSON】 ★★★

    扩展性：不同的子类可以根据具体需求实现自己的 _run 方法
           不同的业务逻辑可能需要不同的数据处理方式和计算方法
    灵活配置：kwargs 参数提供了方法调用时灵活传递参数的能力，可以根据需要在不同情况下调整行为。
    调试和监控：通过 logging.debug 和 set_output 记录输入和输出，可以在调试和生产环境中跟踪程序的行为，并及时发现问题。
    
    def run(self, history, **kwargs):
        logging.debug("{}, history: {}, kwargs: {}".format(self, json.dumps(history, ensure_ascii=False),
                                                              json.dumps(kwargs, ensure_ascii=False)))
        self._param.debug_inputs = []
        try:
            res = self._run(history, **kwargs)
            self.set_output(res)
        except Exception as e:
            self.set_output(pd.DataFrame([{"content": str(e)}]))
            raise e

        return res

    def _run(self, history, **kwargs):
        raise NotImplementedError()

【自定义组件】BaiduFanyi →→ ComponentBase → abc.py 【Python的抽象基类(类似于Java的接口)】
确保某些类在继承抽象基类时遵循特定的协议 这在设计大型系统或者框架时非常有用
★ 接口定义：定义了组件必须实现的方法 => `_run` => 组件执行其任务的地方
★ 组件命名：通过 component_name 类属性，它提供了组件的名称，这有助于在系统中识别和管理组件
★ 运行逻辑：`_run` 方法是组件的核心，它包含了组件的执行逻辑，例如如何使用参数、如何处理输入、如何调用外部API等
★ 输入/输出处理：output → base.py 用于处理组件的输出，确保所有组件的输出格式一致
------------------------------------------------------------------------------
【自定义组件参数】BaiduFanyiParam →→ ComponentParamBase → base.py 【初始化+校验】
定义了百度翻译组件所需的参数，例如 appid、secret_key、trans_type 等，它还实现了一个 check 方法，用于验证参数的有效性。
★ 参数封装：它封装了组件运行所需的所有参数，使得参数管理更加集中方便
★ 参数验证：通过 check 方法，可以确保在组件运行前所有必要的参数都被正确设置，并且是有效的
★ 可扩展性：其他组件的参数类可以继承 ComponentParamBase 并根据需要添加或修改参数

import random
# 抽象基类模块 用于定义组件的基类
from abc import ABC
# 用于发送 HTTP 请求，与 百度翻译 API 交互
import requests
# 基础组件类，于构建此组件的父类，定义组件的基本结构和参数。
from agent.component.base import ComponentBase, ComponentParamBase
# 用于生成百度翻译 API 请求签名（加密哈希）
from hashlib import md5

# 百度翻译组件的参数定义类 继承ComponentParamBase
class BaiduFanyiParam(ComponentParamBase):
    """
    Define the BaiduFanyi component parameters.
    """

    def __init__(self):
        super().__init__()
   # appid 和 secret_key 是百度翻译 API 的 身份验证 信息（需要在百度翻译开放平台获取）
        self.appid = "xxx"
        self.secret_key = "xxx"
   # translate：普通翻译   fieldtranslate：专业领域翻译
        self.trans_type = 'translate'
        self.parameters = []
   # source_lang 翻译的源语言   target_lang 翻译的目标语言
        self.source_lang = 'auto'
        self.target_lang = 'auto'
   # domain 如果使用专业领域翻译，需指定领域（如 finance 表示 金融 领域）
        self.domain = 'finance'
# 检查参数合法性
    def check(self):
    # check_empty()：确保 appid 和 secret_key 不能为空。
        self.check_empty(self.appid, "BaiduFanyi APPID")
        self.check_empty(self.secret_key, "BaiduFanyi Secret Key")
# check_valid_value()：确保 trans_type、source_lang、target_lang、domain 的值在合法选项之内
        self.check_valid_value(self.trans_type, "Translate type", ['translate', 'fieldtranslate'])
# 以下是列举各种语言
        self.check_valid_value(self.source_lang, "Source language",
                               ['auto', 'zh', 'en', 'yue', 'wyw', 'jp', 'kor', 'fra', 'spa', 'th', 'ara', 'ru', 'pt',
                                'de', 'it', 'el', 'nl', 'pl', 'bul', 'est', 'dan', 'fin', 'cs', 'rom', 'slo', 'swe',
                                'hu', 'cht', 'vie'])
        self.check_valid_value(self.target_lang, "Target language",
                               ['auto', 'zh', 'en', 'yue', 'wyw', 'jp', 'kor', 'fra', 'spa', 'th', 'ara', 'ru', 'pt',
                                'de', 'it', 'el', 'nl', 'pl', 'bul', 'est', 'dan', 'fin', 'cs', 'rom', 'slo', 'swe',
                                'hu', 'cht', 'vie'])
# 以下是列举不同专业领域
        self.check_valid_value(self.domain, "Translate field",
                               ['it', 'finance', 'machinery', 'senimed', 'novel', 'academic', 'aerospace', 'wiki',
                                'news', 'law', 'contract'])

# 继承ComponentBase和ABC


# -------------------------------------------------------------------------------- #


class BaiduFanyi(ComponentBase, ABC):
 # 定义组件名称在系统内唯一标识该组件
    component_name = "BaiduFanyi"
    # _run()是组件的 核心执行函数，用于处理翻译请求。
# **kwargs：可以让函数更加灵活，因为它可以接受任意数量的命名参数
# 在函数内部，kwargs 是一个字典，包含了所有传递给函数的额外命名参数。
    def _run(self, history, **kwargs):
'''
self.get_input() 获取输入内容。
如果 content 存在，则用 - 连接多个内容（拼接成单个字符串）。
如果 ans 为空，直接返回 ""
'''
        ans = self.get_input()
        ans = " - ".join(ans["content"]) if "content" in ans else ""
        if not ans:
            return BaiduFanyi.be_output("")

        try:
'''
source_lang 和 target_lang：来源和目标语言。
appid：百度翻译 API 的 应用 ID。
salt：随机数（百度 API 需要此参数）。
secret_key：百度 API 的 密钥
'''
            source_lang = self._param.source_lang
            target_lang = self._param.target_lang
            appid = self._param.appid
            salt = random.randint(32768, 65536)
            secret_key = self._param.secret_key
-------------------------- ☆ 普通翻译API请求 ☆ --------------------------
            if self._param.trans_type == 'translate':
            # md5签名，防止请求被篡改
                sign = md5((appid + ans + salt + secret_key).encode('utf-8')).hexdigest()
     # 发送 HTTP POST 请求 访问 百度翻译 API
                url = 'http://api.fanyi.baidu.com/api/trans/vip/translate?' + 'q=' + ans + '&from=' + source_lang + '&to=' + target_lang + '&appid=' + appid + '&salt=' + salt + '&sign=' + sign
                headers = {"Content-Type": "application/x-www-form-urlencoded"}
            # 解析返回结果
                response = requests.post(url=url, headers=headers).json()

                if response.get('error_code'):
                    BaiduFanyi.be_output("**Error**:" + response['error_msg'])

                return BaiduFanyi.be_output(response['trans_result'][0]['dst'])
-------------------------- ★ 专业翻译API请求 ★ --------------------------
            elif self._param.trans_type == 'fieldtranslate':
                domain = self._param.domain
                sign = md5((appid + ans + salt + domain + secret_key).encode('utf-8')).hexdigest()
                url = 'http://api.fanyi.baidu.com/api/trans/vip/fieldtranslate?' + 'q=' + ans + '&from=' + source_lang + '&to=' + target_lang + '&appid=' + appid + '&salt=' + salt + '&domain=' + domain + '&sign=' + sign
                headers = {"Content-Type": "application/x-www-form-urlencoded"}
                response = requests.post(url=url, headers=headers).json()

                if response.get('error_code'):
                    BaiduFanyi.be_output("**Error**:" + response['error_msg'])

                return BaiduFanyi.be_output(response['trans_result'][0]['dst'])
        # 捕获 所有异常，防止程序崩溃
        except Exception as e:
            BaiduFanyi.be_output("**Error**:" + str(e))

针对于自定义组件参数[ComponentParamBase] 为什么里面是用的Json格式？

• 方便序列化：json.dumps()将对象转换为JSON字符串，可以轻松地保存或传输对象数据。

  base.py
      def __str__(self):
          """
          {
              "component_name": "Begin",
              "params": {}
          }
          """
          # .format()将变量插入到字符串模板中的占位符位置  {{}}转义{}
          return """{{
                "component_name": "{}",
                "params": {},
                "output": {},
                "inputs": {}
            }}""".format(self.component_name,
                       self._param,
  # json.dumps(...): 将获取的 "output" 和 "inputs" 值序列化为 JSON 格式的字符串
  # json.loads(str(self._param)): 将 self._param 转换为字符串，然后将其解析为 JSON 对象
  # self._param 转换为字符串，然后将其解析为 JSON 对象
  # 从 JSON 对象中获取 "output" 键对应的值，如果不存在则返回空字典 {}
                       json.dumps(json.loads(str(self._param)).get("output", {}), ensure_ascii=False),
                       json.dumps(json.loads(str(self._param)).get("inputs", []), ensure_ascii=False)
          )
  

• 方便调试和展示：在 __str__ 方法中返回 JSON 字符串，可以方便地查看对象的内容。尤其是在调试过程中，直接打印出对象的 JSON 格式可以帮助开发人员快速查看对象的状态及其内部数据结构

  # 将对象转换为 JSON 字符串
      def __str__(self):
          return json.dumps(self.as_dict(), ensure_ascii=False)
  # 递归地将对象的属性转换成字典 适用于将对象序列化为 JSON 或进行其他操作
      def as_dict(self):
          def _recursive_convert_obj_to_dict(obj):
              ret_dict = {}
              for attr_name in list(obj.__dict__):
                  if attr_name in [_FEEDED_DEPRECATED_PARAMS, _DEPRECATED_PARAMS, _USER_FEEDED_PARAMS, _IS_RAW_CONF]:
                      continue
                  # get attr
                  attr = getattr(obj, attr_name)
                  if isinstance(attr, pd.DataFrame):
                      ret_dict[attr_name] = attr.to_dict()
                      continue
                  if attr and type(attr).__name__ not in dir(builtins):
                      ret_dict[attr_name] = _recursive_convert_obj_to_dict(attr)
                  else:
                      ret_dict[attr_name] = attr
              return ret_dict
          return _recursive_convert_obj_to_dict(self)
  

• 与大模型交互：当涉及到与大语言模型（如 GPT）或其他机器学习模型的交互时，JSON 格式的数据通常是标准的输入和输出格式。将对象转为 JSON 字符串，能够更容易地将数据传递给模型进行处理或分析，模型通常会接受 JSON 格式的数据进行训练或推理。

如何去运用？

• 数据交换：可以将这个 JSON 字符串用作 API 请求或响应的数据格式。例如，你可以将这个对象作为 HTTP 请求的 body 发送，或者从网络中获取 JSON 格式的数据，然后解析回对象。
• 持久化存储：如果你需要将对象数据持久化到数据库或文件系统，JSON 是一个很好的存储格式。例如，将对象数据存储在文件中或数据库表的 JSON 类型字段中，便于未来读取和操作。
• 配置和参数更新：update 方法中接收到的 conf 参数实际上是一个字典对象，它通过递归的方式更新对象的属性。如果你将对象序列化为 JSON 格式后，可以将 JSON 作为配置文件传递给应用，应用根据该配置动态调整其行为。这使得系统更加灵活和可配置。
• 验证与校验：validate 方法利用存储在 JSON 文件中的规则对对象进行参数验证，这是一种常见的方式来确保输入的数据符合特定的格式或限制。可以通过动态加载配置文件来验证对象的数据，保证数据的合法性。

数据输出(元组)及转换(JSON)

----------------------------------- 输出为元组 -----------------------------------    
    def output(self, allow_partial=True) -> Tuple[str, Union[pd.DataFrame, partial]]:
        o = getattr(self._param, self._param.output_var_name)
        if not isinstance(o, partial):
            if not isinstance(o, pd.DataFrame):
                if isinstance(o, list):
                    return self._param.output_var_name, pd.DataFrame(o)
                if o is None:
                    return self._param.output_var_name, pd.DataFrame()
                return self._param.output_var_name, pd.DataFrame([{"content": str(o)}])
            return self._param.output_var_name, o

        if allow_partial or not isinstance(o, partial):
            if not isinstance(o, partial) and not isinstance(o, pd.DataFrame):
                return pd.DataFrame(o if isinstance(o, list) else [o])
            return self._param.output_var_name, o

        outs = None
        for oo in o():
            if not isinstance(oo, pd.DataFrame):
                # 最终返回的是一个元组
                outs = pd.DataFrame(oo if isinstance(oo, list) else [oo])
            else:
                outs = oo
        return self._param.output_var_name, outs

----------------------------------- 序列化为JSON -----------------------------------
 # 将对象转换为 JSON 字符串
    def __str__(self):
        return json.dumps(self.as_dict(), ensure_ascii=False)

    # 递归地将对象的属性转换成字典 适用于将对象序列化为 JSON 或进行其他操作
    def as_dict(self):
        def _recursive_convert_obj_to_dict(obj):
            ret_dict = {}
            for attr_name in list(obj.__dict__):
                if attr_name in [_FEEDED_DEPRECATED_PARAMS, _DEPRECATED_PARAMS,
                                 _USER_FEEDED_PARAMS, _IS_RAW_CONF]:
                    continue
                # get attr
                attr = getattr(obj, attr_name)
                if isinstance(attr, pd.DataFrame):
                    ret_dict[attr_name] = attr.to_dict()
                    continue
                if attr and type(attr).__name__ not in dir(builtins):
                    ret_dict[attr_name] = _recursive_convert_obj_to_dict(attr)
                else:
                    ret_dict[attr_name] = attr

            return ret_dict

        return _recursive_convert_obj_to_dict(self)
    
--------------------------------------------------------------------------------------
def get_stream_input(self)：
def get_input_elements(self):
    
def get_input(self)：
# 上游组件的输出：它会检查当前组件的上游组件，并尝试从这些组件获取输出数据作为当前的输入
# 如果 _param.query 存在，它会遍历查询参数，并根据参数的类型（如组件 ID）来决定如何获取相应的输入数据
# input后再通过DataFrame转换为元组 再去转换为字典
# 来回来转换是为了满足特定的接口要求或者为了确保数据在不同层之间传输时的兼容性



'''
                           使用Pandas库来处理数据
★ get_input函数: 这个函数首先检查是否有调试输入，如果有，则返回一个包含调试输入内容的DataFrame。然后，它处理查询输入，根据查询中的不同条件（如component_id、value等），调用其他函数获取相应的输入内容，并将这些内容添加到self._param.inputs列表中。最后，它通过pd.concat函数将所有上游组件[结合history]的输出合并成一个DataFrame，并返回。

★ get_input_elements函数: 这个函数根据self._param.query中的查询，构建一个输入元素列表，其中每个元素包含名称、键和值。

★get_stream_input函数: 这个函数处理流式输入，根据组件路径，调用其他函数获取相应的输入内容，并返回。

主要是处理输入参数，并根据不同的条件获取相应的输入内容。它并没有直接接受不同种类类型的入参，而是根据传入的查询参数来获取输入内容。
'''

① agent/component/baidufanyi.py 【后端代码】
② agent/component/_ _ init _ _.py 【初始化】
③ web/src/pages/flow/flow-drawer/index.tsx【表单抽屉组件 UI展示】
④ web/src/pages/flow/form/baidu-fanyi-form/index.tsx 【提供用户界面让用户能配置并提交参数】
⑤ web/src/pages/flow/constant.tsx 【定义一个应用程序中使用的各种图标、常量、枚举、接口、状态和函数】
⑥ web/src/locales/zh.ts 【把所有要显示的搞到zh.ts中(前端组件描述)】
⑦ web/src/pages/agent/constant.tsx【引入组件所需的各种图标、以及操作项的样式、初始值、语言等】
⑧ web/src/pages/agent/form-sheet/use-form-config-map.tsx 【表单配置映射、配置不同操作对应表单组件】
⑨ web/src/pages/agent/form/baidu-fanyi-form/index.tsx 【构建配置表单 】
⑩ web/src/pages/agent/hooks.tsx 【管理图形界面流程图 各种钩子 → 建数据处理流程】

创建动态代理组件 [实战]

📌 实现步骤

1. 后端（Python FastAPI）
   • 创建一个 API 服务器，提供组件注册、查询、运行接口。
   • 组件信息存储在内存（可扩展到数据库）。
   • 组件代码动态执行，支持 exec() 加载。
2. 前端（HTML + JavaScript）
   • 提供一个表单，允许用户输入组件名称、参数、Python 代码。
   • 提交表单后，通过 fetch() 发送 POST 请求给后端。
   • 组件创建成功后，前端可以调用 run 接口测试组件。
3. Postman 测试
   • 先创建组件 POST /api/components/create
   • 再运行组件 POST /api/components/run

1️⃣ 后端代码（FastAPI）

创建 server.py，这个文件用于启动 FastAPI 服务器，并提供 API 端点。

安装依赖（如果未安装 FastAPI 和 Uvicorn）：

pip install fastapi uvicorn pandas

📌 代码（server.py）：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import pandas as pd

app = FastAPI()

# 组件存储（模拟数据库）
components = {}

class ComponentData(BaseModel):
    name: str
    params: dict
    code: str

@app.post("/api/components/create")
def create_component(component: ComponentData):
    """
    创建新的动态组件，存储在内存中，并使用 exec() 加载组件代码
    """
    if component.name in components:
        raise HTTPException(status_code=400, detail="组件已存在")
    
    # 安全地执行代码，存储类定义
    local_vars = {}
    exec(component.code, globals(), local_vars)
    
    if component.name not in local_vars:
        raise HTTPException(status_code=400, detail="代码中必须定义同名类")
    
    components[component.name] = {
        "params": component.params,
        "class": local_vars[component.name]
    }
    return {"message": f"组件 {component.name} 创建成功"}

@app.get("/api/components")
def list_components():
    """
    获取所有注册的组件
    """
    return {"components": list(components.keys())}

class RunRequest(BaseModel):
    component_name: str
    history: list
    params: dict

@app.post("/api/components/run")
def run_component(request: RunRequest):
    """
    运行指定的组件，并返回其输出
    """
    if request.component_name not in components:
        raise HTTPException(status_code=404, detail="组件不存在")

    component_class = components[request.component_name]["class"]
    component_instance = component_class()
    
    # 调用 _run 方法，模拟运行
    try:
        result = component_instance._run(request.history, **request.params)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"组件执行错误: {str(e)}")
    
    return {"result": result.to_dict(orient="records")}

# 运行服务器（可选：手动执行 uvicorn server:app --reload）
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

2️⃣ 前端代码

📌 HTML 页面

📌 代码（index.html）：

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>动态组件管理</title>
</head>
<body>
    <h2>创建动态组件</h2>
    <form id="componentForm">
        <label>组件名称：</label>
        <input type="text" id="componentName" required><br><br>

        <label>参数 (JSON 格式)：</label>
        <textarea id="componentParams" required>{"param1": "value1"}</textarea><br><br>

        <label>Python 代码：</label>
        <textarea id="componentCode" required>
class TestComponent:
    def _run(self, history, **kwargs):
        return pd.DataFrame([{'content': 'Hello, this is a test!'}])
        </textarea><br><br>

        <button type="button" onclick="submitNewComponent()">提交组件</button>
    </form>

    <h2>运行组件</h2>
    <label>组件名称：</label>
    <input type="text" id="runComponentName" required><br><br>
    <button type="button" onclick="runComponent()">运行组件</button>

    <h3>运行结果：</h3>
    <pre id="result"></pre>

    <script src="test.js"></script>
</body>
</html>

📌 JavaScript 代码

📌 代码（test.js）：

const apiBaseUrl = "http://localhost:8000/api";  // 替换为你的服务器地址

function submitNewComponent() {
    const name = document.getElementById("componentName").value;
    const params = document.getElementById("componentParams").value;
    const code = document.getElementById("componentCode").value;

    fetch(`${apiBaseUrl}/components/create`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
            name: name,
            params: JSON.parse(params),
            code: code
        })
    })
    .then(response => response.json())
    .then(data => alert(data.message || JSON.stringify(data)))
    .catch(error => console.error("Error:", error));
}

function runComponent() {
    const componentName = document.getElementById("runComponentName").value;

    fetch(`${apiBaseUrl}/components/run`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
            component_name: componentName,
            history: [],
            params: {}
        })
    })
    .then(response => response.json())
    .then(data => {
        document.getElementById("result").innerText = JSON.stringify(data, null, 2);
    })
    .catch(error => console.error("Error:", error));
}

3️⃣ 使用 Postman 测试

启动服务器后，可以使用 Postman 测试 API。

（1）创建动态组件

• 方法：POST

• URL：http://localhost:8000/api/components/create

• Body（JSON）

  {
    "name": "TestComponent",
    "params": {
      "param1": "value1"
    },
    "code": "class TestComponent:\n    def _run(self, history, **kwargs):\n        return pd.DataFrame([{'content': 'Hello, world!'}])"
  }
  

• 期望结果

  {
    "message": "组件 TestComponent 创建成功"
  }
  

（2）运行组件

• 方法：POST

• URL：http://localhost:8000/api/components/run

• Body（JSON）

  ：

  {
    "component_name": "TestComponent",
    "history": [],
    "params": {}
  }
  

• 期望结果

  ：

  {
    "result": [{"content": "Hello, world!"}]
  }
  

总结

• 后端：FastAPI 实现了组件创建、查询、运行的 API。
• 前端：HTML + JS 提供交互界面，用户可输入参数和代码。
• 测试：可用 Postman 调试，确保组件能被动态创建和运行。

🚀 现在，你可以在 Postman 或前端界面中测试你的动态组件功能了！

纯后端情况如何添加自定义组件呢

这里以我琢磨的JSONPath自定义组件为例

首先要清楚RAGFlow的执行流程 要了解画布的流程 推荐先看一遍RAGFlow执行流程的文档 然后了解一下canvas.py和canvas_app.py的源码 掌握整体流程。

这里最重要的一个思想要明白 现在你只有后端 你没有前端 你在RAGFlow图形化界面拖拽进去的形式已经无法针对于纯后端添加自定义组件了 那怎么办？我们来梳理一下思路 我们想要什么？我们想要图形化界面拖拽过后的整体数据 也就是数据库里的dsl，我们要拿到它 然后用我们自己开发的自定义组件去解析这个大大的JSON。那么这里就涉及到没有前端 你如何去后端去添加自己做的自定义组件！当然常规方法 去图形化界面拖拽已经无法完成这种操作 你只能把希望寄托在数据库

我们来打开数据库 此时你需要一个组件的数据表 component

– rag_flow.component definition

CREATE TABLE component (

id varchar(32) NOT NULL,

create_time bigint DEFAULT NULL,

create_date datetime DEFAULT NULL,

update_time bigint DEFAULT NULL,

update_date datetime DEFAULT NULL,

tenant_id varchar(32) NOT NULL,

module varchar(255) NOT NULL,

created_by varchar(32) NOT NULL,

is_deleted tinyint DEFAULT ‘0’,

PRIMARY KEY (id)

) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

这是创建表的信息 目前我们只需要关注module里面的 我们要把我们自己开发的组件按照那些形式添加进去 我的组件叫:”jsonpath” 那么我的module里面就应该是agent.component.jsonpath 当然你要按照它的风格把组件放在这个包里面。

其次还要注意在这个包里面的__init__.py

把你的组件加进去 “jsonpath”:”JSONPath”

此时并非万事大吉 因为我们目前只有纯后端 没有前端那些 所以回到开头所说 你要自己把你的组件以json的格式手动加入到数据库的dsl中 因为很长所以你需要JSON在线解析格式化验证 - JSON.cn去让它展示的更直观

这是我组件需要的json 你要手动的去添加到dsl 步骤就是先复制数据库的dsl 然后放在解析器里面 手动的去添加自己的组件

“JSONPath:PathProcessor”: {“obj”: {“component_name”: “JSONPath”,”params”: {“output_var_name”: “output”,”jsonpath_expr”: “$.components[*].obj.component_name”,”debug_mode”: true,”query”: [],”inputs”: [],”debug_inputs”: [],”output”: {“content”: {“0”: “”}}}},”downstream”: [“Generate:TwelveClocksSpeak”],”upstream”: []}

这只是我的组件json 你需要在完整的dsl把它插入进去 按照它们形成的格式 这里说明一下 dsl就是你在图形化界面创建agent的时候里面各各组件的相关信息 当然你还要注意自己写的组件类型 是用来干什么的 你才能去做到对应相应的处理 比如说我这个是解析json的 那我就不能去拿上下文组件的输入或输出 正常的流程是连线后你要拿到上游的输出结果 作为你组件的输入结果。我这个则是拿到全局画布canvas_data数据再去进行解析

所以要明白需求才能去更好的融入代码中 经过这几天的学习真的感觉 只有自己一点点的摸索过后才能真正的印象深刻 才能慢慢变得强大！