diff --git a/ddtrace/contrib/celery/utils.py b/ddtrace/contrib/celery/utils.py
index b8c245198b3..e526853f038 100644
--- a/ddtrace/contrib/celery/utils.py
+++ b/ddtrace/contrib/celery/utils.py
@@ -2,6 +2,7 @@
 from typing import Dict
 from weakref import WeakValueDictionary
 
+from ddtrace.contrib.trace_utils import set_flattened_tags
 from ddtrace.span import Span
 
 from .constants import CTX_KEY
@@ -33,27 +34,37 @@
 )
 
 
+def should_skip_context_value(key, value):
+    # type: (str, Any) -> bool
+    # Skip this key if it is not set
+    if value is None or value == "":
+        return True
+
+    # Skip `timelimit` if it is not set (its default/unset value is a
+    # tuple or a list of `None` values
+    if key == "timelimit" and all(_ is None for _ in value):
+        return True
+
+    # Skip `retries` if its value is `0`
+    if key == "retries" and value == 0:
+        return True
+
+    return False
+
+
 def set_tags_from_context(span, context):
     # type: (Span, Dict[str, Any]) -> None
     """Helper to extract meta values from a Celery Context"""
 
+    context_tags = []
     for key, tag_name in TAG_KEYS:
         value = context.get(key)
-
-        # Skip this key if it is not set
-        if value is None or value == "":
-            continue
-
-        # Skip `timelimit` if it is not set (its default/unset value is a
-        # tuple or a list of `None` values
-        if key == "timelimit" and all(_ is None for _ in value):
+        if should_skip_context_value(key, value):
             continue
 
-        # Skip `retries` if its value is `0`
-        if key == "retries" and value == 0:
-            continue
+        context_tags.append((tag_name, value))
 
-        span.set_tag(tag_name, value)
+    set_flattened_tags(span, context_tags)
 
 
 def attach_span(task, task_id, span, is_publish=False):
diff --git a/releasenotes/notes/celery-flatten-context-dicts-bc5c33b72bd72ac2.yaml b/releasenotes/notes/celery-flatten-context-dicts-bc5c33b72bd72ac2.yaml
new file mode 100644
index 00000000000..ffbfbd055b9
--- /dev/null
+++ b/releasenotes/notes/celery-flatten-context-dicts-bc5c33b72bd72ac2.yaml
@@ -0,0 +1,3 @@
+features:
+  - |
+    celery: Enhances context tags containing dictionaries so that their contents are sent as individual tags (issue #4771).
diff --git a/tests/contrib/celery/test_utils.py b/tests/contrib/celery/test_utils.py
index 27824faa0f3..93d1ed21214 100644
--- a/tests/contrib/celery/test_utils.py
+++ b/tests/contrib/celery/test_utils.py
@@ -19,7 +19,7 @@ def test_tags_from_context(self):
         # it should extract only relevant keys
         context = {
             "correlation_id": "44b7f305",
-            "delivery_info": '{"eager": "True"}',
+            "delivery_info": {"eager": "True", "priority": "0", "int_zero": 0},
             "eta": "soon",
             "expires": "later",
             "hostname": "localhost",
@@ -36,7 +36,9 @@ def test_tags_from_context(self):
         metrics = span.get_metrics()
         sentinel = object()
         assert metas["celery.correlation_id"] == "44b7f305"
-        assert metas["celery.delivery_info"] == '{"eager": "True"}'
+        assert metas["celery.delivery_info.eager"] == "True"
+        assert metas["celery.delivery_info.priority"] == "0"
+        assert metrics["celery.delivery_info.int_zero"] == 0
         assert metas["celery.eta"] == "soon"
         assert metas["celery.expires"] == "later"
         assert metas["celery.hostname"] == "localhost"