fix: coderabbit comments on the pr

Author: pira998

examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py

@@ -16,7 +16,7 @@
 from nat.runtime.loader import load_workflow
 
 
-async def create_simple_config():
+async def create_simple_config() -> str:
     """Create a simple workflow config using built-in NAT functions."""
 
     config_content = """
@@ -39,7 +39,7 @@ async def create_simple_config():
         return f.name
 
 
-async def main():
+async def main() -> None:
     """Demonstrate real NAT workflow with cross-workflow observability."""
 
     print("=" * 50)

examples/observability/cross_workflow_tracking/integrated_example.py

@@ -174,7 +174,7 @@ async def create_sample_workflow_configs():
     temp_dir = tempfile.mkdtemp()
 
     # Data validation workflow config
-    validation_config = f"""
+    validation_config = """
 llms:
   mock_llm:
     _type: mock
@@ -193,7 +193,7 @@ async def create_sample_workflow_configs():
         f.write(validation_config.strip())
 
     # Data processing workflow config
-    processing_config = f"""
+    processing_config = """
 llms:
   mock_llm:
     _type: mock
@@ -212,7 +212,7 @@ async def create_sample_workflow_configs():
         f.write(processing_config.strip())
 
     # Analysis workflow config
-    analysis_config = f"""
+    analysis_config = """
 llms:
   mock_llm:
     _type: mock

examples/observability/cross_workflow_tracking/observability_config.yml

@@ -30,10 +30,10 @@ general:
 
 llms:
   nim_llm:
-    _type: nim
-    model_name: meta/llama-3.1-8b-instruct
-    temperature: 0.7
-    max_tokens: 1024
+    _type: nat_test_llm
+    response_seq:
+      - "This is a test response for observability demonstration."
+    delay_ms: 100
 
 workflow:
   _type: chat_completion

external/nat-ui

@@ -13,10 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
+
 from nat.data_models.span import Span
 from nat.observability.processor.processor import Processor
 from nat.utils.type_utils import override
 
+logger = logging.getLogger(__name__)
+
 
 class CrossWorkflowProcessor(Processor[Span, Span]):
     """
@@ -98,8 +102,9 @@ async def process(self, item: Span) -> Span:
                 for key, value in obs_context.custom_attributes.items():
                     item.set_attribute(f"observability.custom.{key}", value)
 
-        except Exception as e:
+        except (AttributeError, KeyError, TypeError, ValueError) as e:
             # If there's any error in processing, log it but don't fail the span
+            logger.warning(f"Error processing cross-workflow observability data: {e}", exc_info=True)
             item.set_attribute("observability.processing_error", str(e))
 
         return item
@@ -158,8 +163,9 @@ async def process(self, item: Span) -> Span:
                     item.set_attribute("relationship.type", "root_workflow")
                     item.set_attribute("relationship.nesting_level", 0)
 
-        except Exception as e:
+        except (AttributeError, IndexError, TypeError) as e:
             # If there's any error in processing, log it but don't fail the span
+            logger.warning(f"Error processing workflow relationship data: {e}", exc_info=True)
             item.set_attribute("relationship.processing_error", str(e))
 
         return item

src/nat/observability/processor/cross_workflow_processor.py

@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
+import sys
 import time
 from typing import TYPE_CHECKING
 from typing import Any
@@ -22,6 +24,8 @@
 
 from nat.observability.context import ObservabilityContext
 
+logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from nat.builder.workflow import Workflow
 
@@ -95,13 +99,15 @@ async def invoke_workflow_with_context(workflow: "Workflow[InputT, StreamingOutp
                 return result
 
         except Exception as e:
-            # Update workflow metadata on failure
+            # Update workflow metadata on failure and log error
             if obs_context:
                 current_workflow = obs_context.get_current_workflow()
                 if current_workflow:
                     current_workflow.end_time = time.time()
                     current_workflow.status = "failed"
                     current_workflow.tags["error"] = str(e)
+
+            logger.error(f"Workflow '{workflow_name}' failed with error: {e}", exc_info=True)
             raise
 
     @staticmethod
@@ -161,13 +167,15 @@ async def invoke_workflow_stream_with_context(
                         current_workflow.status = "completed"
 
         except Exception as e:
-            # Update workflow metadata on failure
+            # Update workflow metadata on failure and log error
             if obs_context:
                 current_workflow = obs_context.get_current_workflow()
                 if current_workflow:
                     current_workflow.end_time = time.time()
                     current_workflow.status = "failed"
                     current_workflow.tags["error"] = str(e)
+
+            logger.error(f"Streaming workflow '{workflow_name}' failed with error: {e}", exc_info=True)
             raise
 
     @staticmethod
@@ -271,11 +279,13 @@ async def invoke_with_steps_and_context(workflow: "Workflow[InputT, StreamingOut
             return result, steps
 
         except Exception as e:
-            # Update workflow metadata on failure
+            # Update workflow metadata on failure and log error
             if obs_context:
                 current_workflow = obs_context.get_current_workflow()
                 if current_workflow:
                     current_workflow.end_time = time.time()
                     current_workflow.status = "failed"
                     current_workflow.tags["error"] = str(e)
+
+            logger.error(f"Workflow with steps '{workflow_name}' failed with error: {e}", exc_info=True)
             raise