Fixing the issue where operators in the tool cannot be used

Author: zhanglongbin

attach/operator/413ddb18-167c-4e22-9177-f873530757ea.png

@@ -52,7 +52,7 @@ def collection_mongo_task(task_uid: str,user_name: str,user_token: str):
             collection_task.task_status = DataSourceTaskStatusEnum.ERROR.value
             insert_datasource_run_task_log_error(task_uid, f"Task with UID {task_uid} has no associated datasource.")
             return False
-        if collection_task.datasource.source_type != DataSourceTypeEnum.MYSQL.value:
+        if collection_task.datasource.source_type != DataSourceTypeEnum.MONGODB.value:
             collection_task.task_status = DataSourceTaskStatusEnum.ERROR.value
             insert_datasource_run_task_log_error(task_uid, f"Task with UID {task_uid} is not a MySQL task.")
             return False

attach/operator/8bf0d6fa-37c9-4c82-a431-726ed4828b44.png

@@ -6,18 +6,18 @@
 from pydantic import Field
 import requests
 
-from ..base_op import OPERATORS, Sample, Selector
+from ..base_op import OPERATORS, Sample, Selector,Param,DataType
 
 
 OP_NAME = 'encode_and_get_nearest_mapper'
 # 编码为嵌入向量
-def get_embeddings(texts: List[str], url: str = "https://ev19h0o3sv7k.space.opencsg.com/embed"):
+def get_embeddings(texts: List[str], model_url):
     """
     Call API service to get text embeddings
 
     Args:
         texts (List[str]): List of texts to encode
-        url (str): API address, defaults to hardcoded address
+        model_url (str): API address, defaults to hardcoded address
 
     Returns:
         List[List[float]]: List of embedding vectors
@@ -32,14 +32,14 @@ def get_embeddings(texts: List[str], url: str = "https://ev19h0o3sv7k.space.open
         "normalize": True
     }
     try:
-        response = requests.post(url, json=payload)
+        response = requests.post(model_url, json=payload)
         response.raise_for_status()  # Raise exception for HTTP errors
         embeddings = response.json()  # List of embeddings
     except requests.RequestException as e:
         raise requests.RequestException(f"Error calling API: {e}")
     return embeddings
 
-def encode_texts(texts: List[str], url: str = "https://ev19h0o3sv7k.space.opencsg.com/embed") -> List[List[float]]:
+def encode_texts(texts: List[str], model_url) -> List[List[float]]:
     """
     Encode multiple texts into embedding vectors
 
@@ -50,7 +50,7 @@ def encode_texts(texts: List[str], url: str = "https://ev19h0o3sv7k.space.opencs
     Returns:
         List[List[float]]: List of embedding vectors
     """
-    return get_embeddings(texts, url=url)
+    return get_embeddings(texts, model_url=model_url)
 
 
 class FaissNearestNeighbour:
@@ -158,6 +158,7 @@ class EncodeAndGetNearestSelector(Selector):
     """Encode texts and find nearest neighbours using Faiss."""
 
     def __init__(self,
+                 model_url: str = "https://ev19h0o3sv7k.space.opencsg.com/embed",
                  *args,
                  **kwargs):
         """
@@ -168,6 +169,7 @@ def __init__(self,
         """
         super().__init__(*args, **kwargs)
         self.first_prompt = []
+        self.model_url = model_url
 
     def process(self, dataset):
         if len(dataset) <= 0:
@@ -176,7 +178,7 @@ def process(self, dataset):
 
         first_prompt_list = dataset["first_prompt"].tolist()
 
-        embeddings = encode_texts(first_prompt_list)
+        embeddings = encode_texts(first_prompt_list,self.model_url)
         dataset['embedding'] = embeddings
 
         nearest_neighbour = FaissNearestNeighbour()
@@ -202,3 +204,10 @@ def sample(cls):
                    "如['What is artificial intelligence?', 'How does machine learning work?']",
             after="数据集增加了embedding、nn_indices和nn_scores字段，包含文本的向量表示和最近邻信息"
         )
+
+    @classmethod
+    @property
+    def init_params(cls):
+        return [
+            Param("model_url", DataType.STRING, {}, "https://ev19h0o3sv7k.space.opencsg.com/embed"),
+        ]

attach/operator/94b91ab5-51fb-4d84-aeef-6271981957f8.png

@@ -21,18 +21,18 @@
 class AnnotateEduTrainBertScorer(Filter):
     def __init__(self,
          auth_token: DataType.STRING = "",
+         model_url: DataType.STRING = "https://esupw2o6m6f4.space.opencsg.com/rerank",
          *args,
          **kwargs):
         super().__init__(*args, **kwargs)
         self.auth_token = auth_token
-
+        self.model_url = model_url
 
     def compute_stats(self, sample, context=False):
         score_field = f"{self.text_key}_score"
         content = sample[self.text_key]
         sample[score_field] = 0
 
-        url = "https://esupw2o6m6f4.space.opencsg.com/rerank"
         # auth_token = "9acc3ea387b5479607bdeb5386af6e3483fbf070"
         data = {
             "query": "What is Deep Learning?",
@@ -44,19 +44,19 @@ def compute_stats(self, sample, context=False):
             "truncate": False,
             "truncation_direction": "right"
         }
-        score = self.get_score_from_model(url,self.auth_token, data)
+        score = self.get_score_from_model(self.model_url,self.auth_token, data)
         if score is not None:
             sample[score_field] = score
         return sample
 
-    def get_score_from_model(self,url, auth_token, data):
+    def get_score_from_model(self,model_url, auth_token, data):
 
         headers = {
             'Content-Type': 'application/json',
             'Authorization': f'Bearer {auth_token}'
         }
 
-        response = requests.post(url, json=data, headers=headers)
+        response = requests.post(model_url, json=data, headers=headers)
 
         if response.status_code == 200:
             try:
@@ -86,4 +86,5 @@ def description(cls):
     def init_params(cls):
         return [
             Param("auth_token", DataType.STRING, {}, ""),
+            Param("model_url", DataType.STRING, {}, "https://esupw2o6m6f4.space.opencsg.com/rerank"),
         ]

data_celery/datasource/mongo/tasks.py

@@ -2,12 +2,11 @@
 # https://github.com/yuyijiong/fineweb-edu-chinese/
 # --------------------------------------------------------
 
-from ..base_op import OPERATORS, Mapper, Sample
+from ..base_op import OPERATORS, Mapper, Sample,Param,DataType
 from ..common import chat_with_model
 
 OP_NAME = 'make_cosmopedia_mapper'
 
-
 @OPERATORS.register_module(OP_NAME)
 class MakeCosmopediaMapper(Mapper):
     """Mapper to generate synthetic tutorial data from seed text samples."""
@@ -66,3 +65,21 @@ def sample(cls):
             'Training your dog to sit is one of the most fundamental commands...'
         )
 
+    @classmethod
+    @property
+    def init_params(cls):
+        return [
+            Param("web_text_max_len", DataType.STRING, {}, 800),
+            Param("model_url", DataType.STRING, {}, "https://euqnoct5ophc.space.opencsg.com/v1/chat/completions"),
+            Param("model", DataType.STRING, {}, "THUDM/LongWriter-glm4-9b"),
+            Param("auth_token", DataType.STRING, {}, "9acc3ea387b5479607bdeb5386af6e3483fbf070"),
+            Param("content", DataType.STRING, {}, '''网页摘录：“{web_text}”。
+以 WikiHow 的风格写一篇长而非常详细的教程，教程与此网页摘录有相关性。
+教程中需要包括对每个步骤的深入解释以及它如何帮助实现预期结果。你可以自由补充其他相关知识。
+确保清晰性和实用性，让读者能够轻松遵循教程完成任务。内容中不应包含广告或涉及隐私的信息。
+不要使用图像。请直接开始撰写教程。
+''')
+        ]
+
+
+

data_engine/ops/edu/encode_and_get_nearest.py

@@ -69,6 +69,7 @@ def __init__(self, tool_defination: Tool_def, params: ExecutedParams):
         )
 
         # whether the model can be accelerated using cuda
+
         _accelerator = self.tool_def.accelerator if self.tool_def.accelerator else None
         if _accelerator is not None:
             self.accelerator = _accelerator
@@ -91,7 +92,7 @@ def run(self):
             # 0. ingest data
             self.tool_def.dataset_path = self.ingester.ingest()
             logger.info(f'Data ingested from {self.tool_def.dataset_path}')
-
+        print('_accelerator', 100 * '*5')
         # 1. data process
         with TRACE_HELPER_TOOL.trace_block(
             "run",
@@ -103,9 +104,11 @@ def run(self):
                 "operation_name": self._name,
             }
         ):
+
             logger.info('Processing tool...')
             tstart = time.time()
             target_path: Path = self.process()
+            print('_accelerator', 100 * '-5')
             tend = time.time()
             logger.info(f'Tool are done in {tend - tstart:.3f}s.')
 

data_engine/ops/edu/pipeline_magpie_zh.py

@@ -123,7 +123,7 @@ async def read_log(id: int,
                    session: Session = Depends(get_sync_session)):
     try:
         log = retreive_log(job_id=id, user_id=user_id,
-                           session=session, isadmin=isadmin, )
+                           session=session, isadmin=isadmin )
         if not log:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
@@ -214,11 +214,26 @@ async def read_task_resource_info(id: int,
 
 @router.post("", response_model=responses.JobCreate, description="Create the dataflow job")
 def create_job(
+
     config:  Union[Tool],
+
+    # config:  Union[Recipe, Tool],
+    # config:  Union[Tool,Recipe],
+    # config: Union[Tool],
+
     user_id: Annotated[str | None, Header(alias="user_id")] = None,
     user_name: Annotated[str | None, Header(alias="user_name")] = None,
     user_token: Annotated[str | None, Header(alias="user_token")] = None
 ):
+    # print(user_id)
+    # print(user_name)
+    # print(user_token)
+    # print(config)
+    if isinstance(config, Recipe):
+        print("匹配到 Recipe 类")
+        # 处理 Recipe 逻辑（如解析 process 字段）
+    elif isinstance(config, Tool):
+        print("匹配到 Tool 类")
     try:
         result = create_new_job(
             job_cfg=config, user_id=user_id, user_name=user_name, user_token=user_token)