添加了主动查询的按钮

app/controller/flow_controller.py

@@ -11,6 +11,7 @@ from app.service.engine.workflow_engine import WorkflowEngine
 from app.service.spec.workflow_loader import load_from_directory
 from app.service.task.task_executor import TaskExecutor
 from app.service.template.template_service import TemplateService
+from app.util.tracking import set_tracking_id
 
 logger = logging.getLogger(__name__)
 
@@ -61,34 +62,35 @@ async def execute(request: Request) -> dict[str, Any]:
     logger.info("[%s] Task accepted, submitting to thread pool", task_id)
 
     def _run():
+        set_tracking_id(task_id)
         t0 = time.time()
         try:
             _record_store.set_processing(task_id)
 
-            logger.info("[%s] Loading template: %s", task_id, template_url)
+            logger.info("Loading template: %s", template_url)
             try:
                 template_path = _template_service.get_template_path(template_url)
                 flow = load_from_directory(template_path)
-                logger.info("[%s] Workflow loaded: version=%s, steps=%s, order=%s",
-                            task_id, flow.version, len(flow.steps), flow.order)
+                logger.info("Workflow loaded: version=%s, steps=%s, order=%s",
+                            flow.version, len(flow.steps), flow.order)
             except Exception as e:
-                logger.error("[%s] Failed to load workflow: %s", task_id, e)
+                logger.error("Failed to load workflow: %s", e)
                 _record_store.set_failed(task_id, f"Failed to load workflow: {e}")
                 return
 
-            logger.info("[%s] Starting workflow execution", task_id)
+            logger.info("Starting workflow execution")
             result = _engine.execute(task_id, flow, inputs)
             elapsed = time.time() - t0
 
             if result.success:
-                logger.info("[%s] Workflow completed successfully in %.2fs", task_id, elapsed)
+                logger.info("Workflow completed successfully in %.2fs", elapsed)
                 _record_store.set_success(task_id, result.outputs)
             else:
-                logger.warning("[%s] Workflow failed in %.2fs: %s", task_id, elapsed, result.error_message)
+                logger.warning("Workflow failed in %.2fs: %s", elapsed, result.error_message)
                 _record_store.set_failed(task_id, result.error_message)
         except Exception as e:
             elapsed = time.time() - t0
-            logger.error("[%s] Workflow execution exception in %.2fs: %s", task_id, elapsed, e, exc_info=True)
+            logger.error("Workflow execution exception in %.2fs: %s", elapsed, e, exc_info=True)
             _record_store.set_failed(task_id, str(e))
 
     _executor.submit(_run)

app/main.py

@@ -14,7 +14,9 @@ from app.repository.local_store import LocalRecordStore
 from app.repository.redis_store import RedisRecordStore
 from app.service.template.template_service import TemplateService
 
-_LOG_FORMAT = "%(asctime)s  %(levelname)-5s  [%(name)s]  %(message)s"
+from app.util.tracking import TrackingFilter
+
+_LOG_FORMAT = "%(asctime)s  %(levelname)-5s  [%(name)s]  %(tracking_id)s%(message)s"
 _LOG_DATEFMT = "%Y-%m-%d %H:%M:%S"
 
 logger = logging.getLogger(__name__)
@@ -29,6 +31,7 @@ def setup_logging() -> None:
         datefmt=_LOG_DATEFMT,
         force=True,
     )
+    logging.getLogger().addFilter(TrackingFilter())
 
 
 @asynccontextmanager

app/repository/record_interface.py

@@ -25,16 +25,16 @@ class RecordInterface(ABC):
         return record_id
 
     def set_processing(self, task_id: str):
-        logger.info("[%s] Status -> processing", task_id)
+        logger.info("Status -> processing")
         self._on_update_status(task_id, "processing", "", None)
 
     def set_failed(self, task_id: str, message: str):
-        logger.warning("[%s] Status -> failed, message=%s", task_id, message)
+        logger.warning("Status -> failed, message=%s", message)
         self._on_update_status(task_id, "failed", message or "", None)
         self._on_publish(task_id)
 
     def set_success(self, task_id: str, outputs: dict[str, Any]):
-        logger.info("[%s] Status -> successes, output_keys=%s", task_id, list(outputs.keys()) if outputs else [])
+        logger.info("Status -> successes, output_keys=%s", list(outputs.keys()) if outputs else [])
         self._on_update_status(task_id, "successes", "", outputs)
         self._on_publish(task_id)
 

app/repository/redis_store.py

@@ -89,7 +89,7 @@ class RedisRecordStore(RecordInterface):
             if record:
                 payload = json.dumps(record.to_dict(), ensure_ascii=False)
                 receivers = self._client.publish(settings.redis_publish_channel, payload)
-                logger.info("[%s] Published to channel '%s', receivers=%d",
-                            task_id, settings.redis_publish_channel, receivers)
+                logger.info("Published to channel '%s', receivers=%d",
+                            settings.redis_publish_channel, receivers)
         except Exception as e:
-            logger.error("[%s] Failed to publish task event: %s", task_id, e)
+            logger.error("Failed to publish task event: %s", e)

app/service/engine/workflow_engine.py

@@ -25,28 +25,28 @@ class WorkflowEngine:
         self, task_id: str, workflow: Workflow, inputs: dict[str, Any]
     ) -> WorkflowResult:
         total_steps = len(workflow.order)
-        logger.info("[%s] ========== Workflow start ==========  version=%s, total_steps=%d",
-                     task_id, workflow.version, total_steps)
+        logger.info("========== Workflow start ==========  version=%s, total_steps=%d",
+                     workflow.version, total_steps)
         wf_t0 = time.time()
 
         ctx = ExecutionContext(inputs)
 
         try:
             workspace = create_task_workspace(task_id, workflow.template_path)
-            logger.info("[%s] Workspace created: %s", task_id, workspace)
+            logger.info("Workspace created: %s", workspace)
         except Exception as e:
-            logger.error("[%s] Failed to create workspace: %s", task_id, e)
+            logger.error("Failed to create workspace: %s", e)
             return WorkflowResult.failure(f"failed: {e}")
 
         for idx, step_id in enumerate(workflow.order, 1):
             step = workflow.steps.get(step_id)
             if step is None:
-                logger.warning("[%s] Step '%s' not found in steps definition, skipping", task_id, step_id)
+                logger.warning("Step '%s' not found in steps definition, skipping", step_id)
                 continue
 
-            logger.info("[%s] ------ Step %d/%d [%s] type=%s start ------",
-                         task_id, idx, total_steps, step.id, step.type)
-            logger.info("[%s] [%s] raw_inputs=%s", task_id, step.id, step.inputs)
+            logger.info("------ Step %d/%d [%s] type=%s start ------",
+                         idx, total_steps, step.id, step.type)
+            logger.info("[%s] raw_inputs=%s", step.id, step.inputs)
 
             step_t0 = time.time()
             try:
@@ -56,31 +56,31 @@ class WorkflowEngine:
                 step_elapsed = time.time() - step_t0
 
                 if result.is_failure():
-                    logger.error("[%s] [%s] FAILED in %.2fs after retries. Error: %s",
-                                  task_id, step_id, step_elapsed, result.error_message)
+                    logger.error("[%s] FAILED in %.2fs after retries. Error: %s",
+                                  step_id, step_elapsed, result.error_message)
                     return WorkflowResult.failure(
                         f"Step {step_id} failed: {result.error_message}"
                     )
             except Exception as e:
                 step_elapsed = time.time() - step_t0
-                logger.error("[%s] [%s] EXCEPTION in %.2fs: %s", task_id, step_id, step_elapsed, e, exc_info=True)
+                logger.error("[%s] EXCEPTION in %.2fs: %s", step_id, step_elapsed, e, exc_info=True)
                 return WorkflowResult.failure(f"Step {step_id} failed: {e}")
 
-            logger.info("[%s] [%s] SUCCESS in %.2fs  outputs=%s",
-                         task_id, step.id, step_elapsed, _shorten(result.outputs))
+            logger.info("[%s] SUCCESS in %.2fs  outputs=%s",
+                         step.id, step_elapsed, _shorten(result.outputs))
             ctx.put_step_result(step_id, result.outputs)
 
-        logger.info("[%s] All steps completed, resolving final outputs ...", task_id)
+        logger.info("All steps completed, resolving final outputs ...")
         outputs: dict[str, Any] = {}
         if workflow.outputs:
             for key, value in workflow.outputs.items():
                 outputs[key] = self._resolver.resolve(value, ctx)
-        logger.info("[%s] Final output_keys=%s", task_id, list(outputs.keys()))
+        logger.info("Final output_keys=%s", list(outputs.keys()))
 
         cleanup_workspace(workspace, task_id)
         total_elapsed = time.time() - wf_t0
-        logger.info("[%s] ========== Workflow end ==========  total_elapsed=%.2fs, success=True",
-                     task_id, total_elapsed)
+        logger.info("========== Workflow end ==========  total_elapsed=%.2fs, success=True",
+                     total_elapsed)
         return WorkflowResult.ok(outputs)
 
     def _execute_step_with_retry(
@@ -101,32 +101,32 @@ class WorkflowEngine:
             )
 
             if attempt > 0:
-                logger.info("[%s] [%s] Retry attempt %d/%d",
-                             task_id, step.id, attempt, max_retries)
+                logger.info("[%s] Retry attempt %d/%d",
+                             step.id, attempt, max_retries)
 
-            logger.info("[%s] [%s] Executing task type=%s, resolved_input_keys=%s",
-                         task_id, step.id, step.type, list(step_inputs.keys()))
+            logger.info("[%s] Executing task type=%s, resolved_input_keys=%s",
+                         step.id, step.type, list(step_inputs.keys()))
 
             attempt_t0 = time.time()
             last_result = task.execute(param)
             attempt_elapsed = time.time() - attempt_t0
 
             if last_result.is_success():
-                logger.info("[%s] [%s] Task returned SUCCESS in %.2fs",
-                             task_id, step.id, attempt_elapsed)
+                logger.info("[%s] Task returned SUCCESS in %.2fs",
+                             step.id, attempt_elapsed)
                 return last_result
 
             if attempt < max_retries:
                 logger.warning(
-                    "[%s] [%s] Task returned FAILURE in %.2fs (attempt %d/%d), will retry. Error: %s",
-                    task_id, step.id, attempt_elapsed,
+                    "[%s] Task returned FAILURE in %.2fs (attempt %d/%d), will retry. Error: %s",
+                    step.id, attempt_elapsed,
                     attempt + 1, max_retries + 1, last_result.error_message,
                 )
                 time.sleep(1)
             else:
                 logger.error(
-                    "[%s] [%s] Task returned FAILURE in %.2fs (final attempt %d/%d). Error: %s",
-                    task_id, step.id, attempt_elapsed,
+                    "[%s] Task returned FAILURE in %.2fs (final attempt %d/%d). Error: %s",
+                    step.id, attempt_elapsed,
                     attempt + 1, max_retries + 1, last_result.error_message,
                 )
 

app/util/task_util.py

@@ -11,11 +11,11 @@ BASE_DIR = "data/data"
 
 def create_task_workspace(task_id: str, template_path: str) -> str:
     workspace = os.path.join(BASE_DIR, task_id)
-    logger.info("[%s] Creating workspace: %s  (from template: %s)", task_id, workspace, template_path)
+    logger.info("Creating workspace: %s  (from template: %s)", workspace, template_path)
     t0 = time.time()
     copy_directory(template_path, workspace)
     elapsed = time.time() - t0
-    logger.info("[%s] Workspace ready in %.2fs", task_id, elapsed)
+    logger.info("Workspace ready in %.2fs", elapsed)
     return workspace
 
 
@@ -24,8 +24,8 @@ def cleanup_workspace(workspace: str, task_id: str):
         return
     try:
         if os.path.isdir(workspace):
-            logger.info("[%s] Cleaning up workspace: %s", task_id, workspace)
+            logger.info("Cleaning up workspace: %s", workspace)
             delete_directory(workspace)
-            logger.info("[%s] Workspace cleaned up", task_id)
+            logger.info("Workspace cleaned up")
     except Exception as e:
-        logger.error("[%s] Error cleaning up workspace %s: %s", task_id, workspace, e)
+        logger.error("Error cleaning up workspace %s: %s", workspace, e)

app/util/tracking.py

@@ -0,0 +1,21 @@
+import contextvars
+import logging
+
+_tracking_id: contextvars.ContextVar[str] = contextvars.ContextVar("tracking_id", default="")
+
+
+def set_tracking_id(task_id: str) -> contextvars.Token:
+    return _tracking_id.set(task_id)
+
+
+def get_tracking_id() -> str:
+    return _tracking_id.get()
+
+
+class TrackingFilter(logging.Filter):
+    """Logging filter that injects the current tracking ID into every log record."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        tid = _tracking_id.get()
+        record.tracking_id = f"[{tid}] " if tid else ""
+        return True