kombu.exceptions.encodeerror：类型集的对象在芹菜中不可用JSON序列化

Question

我正在使用快速的API，在其中使用芹菜进行任务队列预测。每当我进行预测并试图返回响应时，我都会收到500未记录的错误：内部服务器错误并从Docker Log中收到错误

[2022-04-18 10:49:51,965: INFO/MainProcess] Connected to redis://redis:6379/0
[2022-04-18 10:49:51,969: INFO/MainProcess] mingle: searching for neighbors
[2022-04-18 10:49:52,980: INFO/MainProcess] mingle: all alone
[2022-04-18 10:49:53,044: INFO/MainProcess] celery@809b6efd8550 ready.
[2022-04-18 10:49:55,704: INFO/MainProcess] Task create_task[efcc7e7e-4fea-4c2d-b135-8aa546418327] received
[2022-04-18 10:50:05,748: ERROR/ForkPoolWorker-8] Task create_task[efcc7e7e-4fea-4c2d-b135-8aa546418327] raised unexpected: EncodeError(TypeError('Object of type set is not JSON serializable'))
Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 39, in _reraise_errors
yield
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 210, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 68, in dumps
return _dumps(s, cls=cls or _default_encoder,
File "/usr/local/lib/python3.8/json/__init__.py", line 234, in dumps
return cls(
File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
return _iterencode(o, 0)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 58, in default
return super().default(o)
File "/usr/local/lib/python3.8/json/encoder.py", line 179, in default
raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type set is not JSON serializable
During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/celery/app/trace.py", line 517, in trace_task
mark_as_done(
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 161, in mark_as_done
self.store_result(task_id, result, state, request=request)
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 482, in store_result
self._store_result(task_id, result, state, traceback,
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 909, in _store_result
self._set_with_state(self.get_key_for_task(task_id), self.encode(meta), state)
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 372, in encode
_, _, payload = self._encode(data)
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 376, in _encode
return dumps(data, serializer=self.serializer)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 210, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/contextlib.py", line 131, in __exit__
self.gen.throw(type, value, traceback)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 43, in _reraise_errors
reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/usr/local/lib/python3.8/site-packages/kombu/exceptions.py", line 21, in reraise
raise value.with_traceback(tb)
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 39, in _reraise_errors
yield
File "/usr/local/lib/python3.8/site-packages/kombu/serialization.py", line 210, in dumps
payload = encoder(data)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 68, in dumps
return _dumps(s, cls=cls or _default_encoder,
File "/usr/local/lib/python3.8/json/__init__.py", line 234, in dumps
return cls(
File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
chunks = self.iterencode(o, _one_shot=True)
File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
return _iterencode(o, 0)
File "/usr/local/lib/python3.8/site-packages/kombu/utils/json.py", line 58, in default
return super().default(o)
File "/usr/local/lib/python3.8/json/encoder.py", line 179, in default
raise TypeError(f'Object of type {o.__class__.__name__} '
kombu.exceptions.EncodeError: Object of type set is not JSON serializable

都不知道为什么我会遇到此错误。

我的worker.py

@celery.task(name='create_task')
def create_task(solute, solvent):
    time.sleep(10)
    return {'Harsha'}

main.py

@app.get(api_names[0])
async def predict(solute,solvent):
    task = create_task.delay(solute,solvent)
    return JSONResponse({'task_id': task.id})

@app.get('/predict_solubility/{task_id}')
async def post(task_id):
    task_result = AsyncResult(task_id)
    result = {
        'task_id': task_id,
        'task_status': task_result.status,
        'task_result' : task_result
    }
    return result

预测

response = {}
async def predictions(solute, solvent):
    response.clear()
    m = Chem.MolFromSmiles(solute,sanitize=False)
    n = Chem.MolFromSmiles(solvent,sanitize=False)
    if (m == None or n == None):
      response['predictions']= 'invalid SMILES'
      print('invalid SMILES')
    else:
      mol = Chem.MolFromSmiles(solute)
      mol = Chem.AddHs(mol)
      solute = Chem.MolToSmiles(mol)
      solute_graph = get_graph_from_smile(solute)
      mol = Chem.MolFromSmiles(solvent)
      mol = Chem.AddHs(mol)
      solvent = Chem.MolToSmiles(mol)
      solvent_graph = get_graph_from_smile(solvent)
      delta_g, interaction_map =  model([solute_graph.to(device), solvent_graph.to(device)])
      interaction_map_one = torch.trunc(interaction_map)
      response["interaction_map"] = (interaction_map_one.detach().numpy()).tolist()
      response["predictions"] = delta_g.item()

结果的屏幕截图

Answer 1

要解释错误消息的实际含义：没有数据类型与JSON中的集合 - IE Python的set（）无法在JSON中表示，因此，当任务结果序列化与JSON序列化时返回，将丢弃错误。

如果您在任务中看到代码，则返回一组：

def create_task(solute, solvent):
    time.sleep(10)
    return {'Harsha'}

{value}语法在Python中创建一个集。您可能已经以为您返回了字典，但是要么需要为空，要么至少有一个键：值对要创建为dict（否则，创建了一个集合 - 如果只有一组 - 单个值或多个值，没有密钥）。

return {'result': 'Harsha'}

..代替返回字典。