Merge pull request #281 from PrefectHQ/flow

Improve debugging options

pyproject.toml

@@ -111,5 +111,7 @@ env = [
     # use 4o-mini for tests by default
     'D:CONTROLFLOW_LLM_MODEL=openai/gpt-4o-mini',
     'D:CONTROLFLOW_TOOLS_VERBOSE=1',
+    'D:CONTROLFLOW_PRETTY_PRINT_AGENT_EVENTS=0',
+    'D:CONTROLFLOW_LOG_LEVEL=DEBUG',
     'D:PREFECT_LOGGING_LEVEL=DEBUG',
 ]

src/controlflow/agents/agent.py

@@ -236,6 +236,9 @@ def _run_model(
 
         model = self.get_model(tools=tools)
 
+        if controlflow.settings.log_all_messages:
+            logger.debug(f"Input messages: {messages}")
+
         if stream:
             response = None
             for delta in model.stream(messages):
@@ -251,6 +254,9 @@ def _run_model(
 
         yield AgentMessage(agent=self, message=response)
 
+        if controlflow.settings.log_all_messages:
+            logger.debug(f"Response: {response}")
+
         for tool_call in response.tool_calls + response.invalid_tool_calls:
             yield ToolCallEvent(agent=self, tool_call=tool_call)
             result = handle_tool_call(tool_call, tools=tools)
@@ -271,6 +277,9 @@ async def _run_model_async(
 
         model = self.get_model(tools=tools)
 
+        if controlflow.settings.log_all_messages:
+            logger.debug(f"Input messages: {messages}")
+
         if stream:
             response = None
             async for delta in model.astream(messages):
@@ -286,6 +295,9 @@ async def _run_model_async(
 
         yield AgentMessage(agent=self, message=response)
 
+        if controlflow.settings.log_all_messages:
+            logger.debug(f"Response: {response}")
+
         for tool_call in response.tool_calls + response.invalid_tool_calls:
             yield ToolCallEvent(agent=self, tool_call=tool_call)
             result = await handle_tool_call_async(tool_call, tools=tools)

src/controlflow/orchestration/orchestrator.py

@@ -6,6 +6,7 @@
 import controlflow
 from controlflow.agents.agent import Agent
 from controlflow.events.base import Event
+from controlflow.events.events import AgentMessageDelta, OrchestratorMessage
 from controlflow.events.message_compiler import MessageCompiler
 from controlflow.flows import Flow
 from controlflow.instructions import get_instructions
@@ -61,7 +62,7 @@ def _validate_handlers(cls, v):
         """
         from controlflow.orchestration.print_handler import PrintHandler
 
-        if v is None and controlflow.settings.enable_print_handler:
+        if v is None and controlflow.settings.pretty_print_agent_events:
             v = [PrintHandler()]
         return v or []
 
@@ -72,7 +73,8 @@ def handle_event(self, event: Event):
         Args:
             event (Event): The event to handle.
         """
-        logger.debug(f"Handling event: {event}")
+        if not isinstance(event, AgentMessageDelta):
+            logger.debug(f"Handling event: {repr(event)}")
         for handler in self.handlers:
             handler.handle(event)
         if event.persist:
@@ -138,7 +140,15 @@ def _run_turn(self, max_calls_per_turn: Optional[int] = None):
         self.turn_strategy.begin_turn()
 
         for task in self.get_tasks("assigned"):
-            task.mark_running()
+            if not task.is_running():
+                task.mark_running()
+                self.flow.add_events(
+                    [
+                        OrchestratorMessage(
+                            content=f"Starting task {task.name} (ID {task.id}) with objective: {task.objective}"
+                        )
+                    ]
+                )
 
         calls = 0
         while not self.turn_strategy.should_end_turn():
@@ -176,7 +186,15 @@ async def _run_turn_async(self, max_calls_per_turn: Optional[int] = None):
         self.turn_strategy.begin_turn()
 
         for task in self.get_tasks("assigned"):
-            task.mark_running()
+            if not task.is_running():
+                task.mark_running()
+                self.flow.add_events(
+                    [
+                        OrchestratorMessage(
+                            content=f"Starting task {task.name} (ID {task.id}) with objective: {task.objective}"
+                        )
+                    ]
+                )
 
         calls = 0
         while not self.turn_strategy.should_end_turn():

src/controlflow/settings.py

@@ -35,10 +35,23 @@ class Settings(ControlFlowSettings):
 
     # ------------ display and logging settings ------------
 
+    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = Field(
+        default="INFO",
+        description="The log level for ControlFlow.",
+    )
     log_prints: bool = Field(
         default=False,
         description="Whether to log workflow prints to the Prefect logger by default.",
     )
+    log_all_messages: bool = Field(
+        default=False,
+        description="If True, all LLM messages will be logged at the debug level.",
+    )
+    pretty_print_agent_events: bool = Field(
+        default=True,
+        description="If True, a PrintHandler will be enabled and automatically "
+        "pretty-print agent events. Note that this may interfere with logging.",
+    )
 
     # ------------ orchestration settings ------------
     orchestrator_max_turns: Optional[int] = Field(
@@ -63,24 +76,13 @@ class Settings(ControlFlowSettings):
         100_000, description="The maximum number of tokens to send to an LLM."
     )
 
-    # ------------ Flow visualization settings ------------
-
-    enable_print_handler: bool = Field(
-        default=True,
-        description="If True, the PrintHandler will be enabled.",
-    )
+    # ------------ Debug settings ------------
 
-    enable_experimental_tui: bool = Field(
-        default=False,
-        description="If True, the experimental TUI will be enabled. If False, the TUI will be disabled.",
-    )
-    run_tui_headless: bool = Field(
+    debug_messages: bool = Field(
         default=False,
-        description="If True, the experimental TUI will run in headless mode, which is useful for debugging.",
+        description="If True, all messages will be logged at the debug level.",
     )
 
-    # ------------ Debug settings ------------
-
     tools_raise_on_error: bool = Field(
         default=False,
         description="If True, an error in a tool call will raise an exception.",
@@ -90,6 +92,17 @@ class Settings(ControlFlowSettings):
         default=True, description="If True, tools will log additional information."
     )
 
+    # ------------ experimental settings ------------
+
+    enable_experimental_tui: bool = Field(
+        default=False,
+        description="If True, the experimental TUI will be enabled. If False, the TUI will be disabled.",
+    )
+    run_tui_headless: bool = Field(
+        default=False,
+        description="If True, the experimental TUI will run in headless mode, which is useful for debugging.",
+    )
+
     # ------------ Prefect settings ------------
     #
     # Default settings for Prefect when used with ControlFlow. They can be
@@ -110,6 +123,13 @@ def _validate_home_path(cls, v: Union[str, Path]) -> Path:
             v.mkdir(parents=True, exist_ok=True)
         return v
 
+    @model_validator(mode="after")
+    def set_log_level(self):
+        from controlflow.utilities.logging import setup_logging
+
+        setup_logging(level=self.log_level)
+        return self
+
     @model_validator(mode="after")
     def _apply_prefect_settings(self):
         """

src/controlflow/utilities/logging.py

@@ -2,6 +2,8 @@
 from functools import lru_cache
 from typing import Optional
 
+import controlflow
+
 
 @lru_cache()
 def get_logger(name: Optional[str] = None) -> logging.Logger:
@@ -41,6 +43,15 @@ def get_logger(name: Optional[str] = None) -> logging.Logger:
     return logger
 
 
+def setup_logging(level: Optional[str] = None) -> None:
+    logger = get_logger()
+
+    if level is not None:
+        logger.setLevel(level)
+    else:
+        logger.setLevel(controlflow.settings.log_level)
+
+
 def deprecated(message: str, version: str):
     """
     Decorator to mark a function as deprecated.

tests/conftest.py

@@ -9,7 +9,9 @@
 @pytest.fixture(autouse=True, scope="session")
 def temp_controlflow_settings():
     with temporary_settings(
-        enable_print_handler=False,
+        pretty_print_agent_events=False,
+        log_all_messages=True,
+        log_level="DEBUG",
         orchestrator_max_turns=10,
         orchestrator_max_calls_per_turn=10,
     ):

tests/test_settings.py

@@ -49,7 +49,7 @@ def test_import_without_default_api_key_warns_but_does_not_fail(monkeypatch, cap
             caplog.clear()
 
             # Import the library
-            with caplog.at_level("WARNING"):
+            with caplog.at_level("DEBUG"):
                 # Reload the library to apply changes
                 defaults_module = importlib.import_module("controlflow.defaults")
                 importlib.reload(defaults_module)

tests/utilities/test_testing.py

@@ -31,17 +31,19 @@ def test_record_task_events(default_fake_llm):
     with record_events() as events:
         task.run()
 
-    assert events[0].event == "agent-message"
-    assert response == events[0].ai_message
+    assert events[0].event == "orchestrator-message"
 
-    assert events[2].event == "tool-result"
-    assert events[2].tool_call == {
+    assert events[1].event == "agent-message"
+    assert response == events[1].ai_message
+
+    assert events[3].event == "tool-result"
+    assert events[3].tool_call == {
         "name": "mark_task_12345_successful",
         "args": {"result": "Hello!"},
         "id": "call_ZEPdV8mCgeBe5UHjKzm6e3pe",
         "type": "tool_call",
     }
-    assert events[2].tool_result.model_dump() == dict(
+    assert events[3].tool_result.model_dump() == dict(
         tool_call_id="call_ZEPdV8mCgeBe5UHjKzm6e3pe",
         str_result='Task 12345 ("say hello") marked successful.',
         is_error=False,