feat(subtasks): segment trajectories into subtasks before tip generation

altk_evolve/config/evolve.py

@@ -8,6 +8,7 @@ class EvolveConfig(BaseSettings):
     namespace_id: str = "evolve"
     settings: BaseSettings | None = None
     clustering_threshold: float = 0.80
+    segmentation_enabled: bool = True
 
 
 # to reload settings call evolve_config.__init__()

altk_evolve/frontend/mcp/mcp_server.py

@@ -193,27 +193,29 @@ def save_trajectory(trajectory_data: str, task_id: str | None = None) -> list[Re
         entities=entities,
         enable_conflict_resolution=False,
     )
-    result = generate_tips(messages)
-
-    if result.tips:
+    results = generate_tips(messages)
+
+    tip_entities = [
+        Entity(
+            type="guideline",
+            content=tip.content,
+            metadata={
+                "category": tip.category,
+                "rationale": tip.rationale,
+                "trigger": tip.trigger,
+                "implementation_steps": tip.implementation_steps,
+                "task_description": result.task_description,
+                "source_task_id": task_id,
+                "creation_mode": "auto-mcp",
+            },
+        )
+        for result in results
+        for tip in result.tips
+    ]
+    if tip_entities:
         get_client().update_entities(
             namespace_id=evolve_config.namespace_id,
-            entities=[
-                Entity(
-                    type="guideline",
-                    content=tip.content,
-                    metadata={
-                        "category": tip.category,
-                        "rationale": tip.rationale,
-                        "trigger": tip.trigger,
-                        "implementation_steps": tip.implementation_steps,
-                        "task_description": result.task_description,
-                        "source_task_id": task_id,
-                        "creation_mode": "auto-mcp",
-                    },
-                )
-                for tip in result.tips
-            ],
+            entities=tip_entities,
             enable_conflict_resolution=True,
         )
 

altk_evolve/llm/tips/prompts/segment_trajectory.jinja2

@@ -0,0 +1,44 @@
+Analyze this AI agent trajectory and segment it into logical subtasks.
+
+# Trajectory
+**Total Steps:** {{num_steps}}
+
+{{trajectory_summary}}
+
+# Your Task
+Identify the distinct logical phases of work the agent performed. For each phase, produce a generalized description that would apply to *any* agent performing a similar operation — strip out specific values, user names, IDs, and task-specific details.
+
+**Good description:** "Authenticate with a web service using stored credentials"
+**Bad description:** "Login to Venmo using tr_solo@gmail.com"
+
+**Guidelines:**
+- Group tightly related steps together (e.g., all authentication steps form one subtask)
+- Separate distinct logical phases (discovery, authentication, data retrieval, computation, output)
+- `start_step` and `end_step` are the step numbers shown in the trajectory above (inclusive)
+- Prefer contiguous, non-overlapping ranges where the trajectory allows it
+- A single step can be its own subtask if it represents a clearly distinct phase
+- `purpose` should describe what this subtask achieves, not how it does it
+
+{% if not constrained_decoding_supported %}
+**Output Format (JSON):**
+```json
+{
+    "subtasks": [
+        {
+            "generalized_description": "Authenticate with a web service using stored credentials",
+            "start_step": 1,
+            "end_step": 4,
+            "purpose": "Obtain an access token to make authorized API calls"
+        },
+        {
+            "generalized_description": "Retrieve and filter a list of records",
+            "start_step": 5,
+            "end_step": 9,
+            "purpose": "Fetch the target dataset and apply the required filters"
+        }
+    ]
+}
+```
+
+Return ONLY the JSON object, no other text.
+{% endif %}

altk_evolve/llm/tips/segmentation.py

@@ -0,0 +1,88 @@
+import json
+import logging
+from json import JSONDecodeError
+from pathlib import Path
+
+import litellm
+from jinja2 import Template
+from litellm import completion, get_supported_openai_params, supports_response_schema
+from pydantic import ValidationError
+
+from altk_evolve.config.llm import llm_settings
+from altk_evolve.schema.tips import SegmentationResponse, SubtaskSegment
+from altk_evolve.utils.utils import clean_llm_response
+
+logger = logging.getLogger(__name__)
+
+_SEGMENT_TEMPLATE = Template((Path(__file__).parent / "prompts/segment_trajectory.jinja2").read_text())
+
+
+def segment_trajectory(messages: list[dict]) -> list[SubtaskSegment]:
+    """Segment a trajectory into logical subtasks with generalized descriptions.
+
+    Returns an empty list on failure — callers fall back to full-trajectory tip generation.
+    """
+    # Import here to avoid circular import (tips.py imports this module)
+    from altk_evolve.llm.tips.tips import parse_openai_agents_trajectory
+
+    trajectory_data = parse_openai_agents_trajectory(messages)
+
+    supported_params = get_supported_openai_params(
+        model=llm_settings.tips_model,
+        custom_llm_provider=llm_settings.custom_llm_provider,
+    )
+    supports_response_format = supported_params and "response_format" in supported_params
+    response_schema_enabled = supports_response_schema(
+        model=llm_settings.tips_model,
+        custom_llm_provider=llm_settings.custom_llm_provider,
+    )
+    constrained_decoding_supported = bool(supports_response_format and response_schema_enabled)
+
+    prompt = _SEGMENT_TEMPLATE.render(
+        trajectory_summary=trajectory_data["trajectory_summary"],
+        num_steps=trajectory_data["num_steps"],
+        constrained_decoding_supported=constrained_decoding_supported,
+    )
+
+    litellm.enable_json_schema_validation = constrained_decoding_supported
+
+    last_error: Exception | None = None
+    for attempt in range(3):
+        try:
+            if constrained_decoding_supported:
+                clean_response = (
+                    completion(
+                        model=llm_settings.tips_model,
+                        messages=[{"role": "user", "content": prompt}],
+                        response_format=SegmentationResponse,
+                        custom_llm_provider=llm_settings.custom_llm_provider,
+                    )
+                    .choices[0]
+                    .message.content
+                )
+            else:
+                raw = (
+                    completion(
+                        model=llm_settings.tips_model,
+                        messages=[{"role": "user", "content": prompt}],
+                        custom_llm_provider=llm_settings.custom_llm_provider,
+                    )
+                    .choices[0]
+                    .message.content
+                )
+                clean_response = clean_llm_response(raw)
+
+            if not clean_response:
+                logger.debug(f"Segmentation attempt {attempt + 1}: empty response")
+                continue
+
+            subtasks = SegmentationResponse.model_validate(json.loads(clean_response)).subtasks
+            return subtasks
+
+        except (JSONDecodeError, ValidationError) as e:
+            logger.debug(f"Segmentation attempt {attempt + 1} failed: {e}")
+            last_error = e
+            continue
+
+    logger.warning(f"Failed to segment trajectory after 3 attempts: {last_error}")
+    return []

altk_evolve/llm/tips/tips.py

@@ -8,13 +8,16 @@
 from litellm import completion, get_supported_openai_params, supports_response_schema
 from pydantic import ValidationError
 
+from altk_evolve.config.evolve import evolve_config
 from altk_evolve.config.llm import llm_settings
 from altk_evolve.schema.exceptions import EvolveException
 from altk_evolve.schema.tips import DEFAULT_TASK_DESCRIPTION, TipGenerationResponse, TipGenerationResult
 from altk_evolve.utils.utils import clean_llm_response
 
 logger = logging.getLogger(__name__)
 
+_GENERATE_TIPS_TEMPLATE = Template((Path(__file__).parent / "prompts/generate_tips.jinja2").read_text())
+
 
 def parse_openai_agents_trajectory(messages: list[dict]) -> dict:
     """
@@ -26,6 +29,7 @@ def parse_openai_agents_trajectory(messages: list[dict]) -> dict:
         - agent_steps: List of agent reasoning/actions
         - function_calls: List of tool/function calls made
         - num_steps: Total number of agent actions
+        - steps_list: Individual formatted step strings (before joining), for subtask slicing
     """
     agent_steps: list[dict[str, str | dict]] = []
     function_calls: list[dict[str, str | dict]] = []
@@ -80,7 +84,7 @@ def parse_openai_agents_trajectory(messages: list[dict]) -> dict:
                 # Skip empty assistant messages (common from tool-calling patterns)
                 continue
 
-    steps_text = []
+    steps_list = []
     for i, step in enumerate(agent_steps[:50], 1):
         step_type = step["type"]
         content = step["content"]
@@ -89,38 +93,30 @@ def parse_openai_agents_trajectory(messages: list[dict]) -> dict:
             content = content[:2000] + "..."
 
         if step_type == "reasoning":
-            steps_text.append(f"**Step {i} - Reasoning:**\n{content}")
+            steps_list.append(f"**Step {i} - Reasoning:**\n{content}")
         elif step_type == "action":
-            steps_text.append(f"**Step {i} - Action:**\n{content}")
-        elif step_type == "observation":
-            steps_text.append(f"**Step {i} - Observation:**\n{content}")
+            steps_list.append(f"**Step {i} - Action:**\n{content}")
 
     return {
         "task_instruction": task_instruction or DEFAULT_TASK_DESCRIPTION,
-        "trajectory_summary": "\n\n".join(steps_text),
+        "trajectory_summary": "\n\n".join(steps_list),
+        "steps_list": steps_list,
         "function_calls": function_calls,
-        "num_steps": len([s for s in agent_steps if s["type"] in ["action", "reasoning"]]),
+        "num_steps": len([s for s in agent_steps[:50] if s["type"] in ["action", "reasoning"]]),
     }
 
 
-def generate_tips(messages: list[dict]) -> TipGenerationResult:
-    prompt_file = Path(__file__).parent / "prompts/generate_tips.jinja2"
-    supported_params = get_supported_openai_params(
-        model=llm_settings.tips_model,
-        custom_llm_provider=llm_settings.custom_llm_provider,
-    )
-    supports_response_format = supported_params and "response_format" in supported_params
-    response_schema_enabled = supports_response_schema(
-        model=llm_settings.tips_model,
-        custom_llm_provider=llm_settings.custom_llm_provider,
-    )
-    constrained_decoding_supported = supports_response_format and response_schema_enabled
-    trajectory_data = parse_openai_agents_trajectory(messages)
-    task_description = trajectory_data["task_instruction"]
-    prompt = Template(prompt_file.read_text()).render(
+def _generate_tips_for_segment(
+    task_description: str,
+    trajectory_slice: str,
+    num_steps: int,
+    constrained_decoding_supported: bool,
+) -> TipGenerationResult:
+    """Generate tips for a single trajectory slice (full or subtask)."""
+    prompt = _GENERATE_TIPS_TEMPLATE.render(
         task_instruction=task_description,
-        num_steps=trajectory_data["num_steps"],
-        trajectory_summary=trajectory_data["trajectory_summary"],
+        num_steps=num_steps,
+        trajectory_summary=trajectory_slice,
         constrained_decoding_supported=constrained_decoding_supported,
     )
 
@@ -138,7 +134,7 @@ def generate_tips(messages: list[dict]) -> TipGenerationResult:
         )
     else:
         litellm.enable_json_schema_validation = False
-        response = (
+        raw = (
             completion(
                 model=llm_settings.tips_model,
                 messages=[{"role": "user", "content": prompt}],
@@ -147,7 +143,8 @@ def generate_tips(messages: list[dict]) -> TipGenerationResult:
             .choices[0]
             .message.content
         )
-        clean_response = clean_llm_response(response)
+        clean_response = clean_llm_response(raw)
+
     if not clean_response:
         logger.warning(f"LLM returned empty response for tip generation. Model: {llm_settings.tips_model}")
         return TipGenerationResult(tips=[], task_description=task_description)
@@ -160,3 +157,68 @@ def generate_tips(messages: list[dict]) -> TipGenerationResult:
     except ValidationError as e:
         logger.warning(f"Failed to validate LLM tip generation response: {e}. Response: {repr(clean_response[:500])}")
         return TipGenerationResult(tips=[], task_description=task_description)
+
+
+def generate_tips(messages: list[dict]) -> list[TipGenerationResult]:
+    """Generate tips from a trajectory, optionally segmented into subtasks.
+
+    When segmentation is enabled (EVOLVE_SEGMENTATION_ENABLED=true, the default),
+    the trajectory is first segmented into logical subtasks. Tips are then generated
+    per subtask and each result carries the subtask's generalized description as
+    task_description — giving downstream clustering much more precise signal than
+    the raw first user message.
+
+    Returns a list with one TipGenerationResult per subtask (or one for the full
+    trajectory when segmentation is disabled or produces fewer than 2 subtasks).
+    """
+    supported_params = get_supported_openai_params(
+        model=llm_settings.tips_model,
+        custom_llm_provider=llm_settings.custom_llm_provider,
+    )
+    supports_response_format = supported_params and "response_format" in supported_params
+    response_schema_enabled = supports_response_schema(
+        model=llm_settings.tips_model,
+        custom_llm_provider=llm_settings.custom_llm_provider,
+    )
+    constrained_decoding_supported = bool(supports_response_format and response_schema_enabled)
+
+    trajectory_data = parse_openai_agents_trajectory(messages)
+    task_instruction = trajectory_data["task_instruction"]
+    steps_list: list[str] = trajectory_data["steps_list"]
+    n_steps = len(steps_list)
+
+    subtasks = []
+    if evolve_config.segmentation_enabled:
+        from altk_evolve.llm.tips.segmentation import segment_trajectory  # avoid circular import
+
+        try:
+            subtasks = segment_trajectory(messages)
+        except Exception as e:
+            logger.warning(f"Trajectory segmentation failed, falling back to full trajectory: {e}")
+            subtasks = []
+
+    if len(subtasks) >= 2:
+        results = []
+        for subtask in subtasks:
+            start = max(0, subtask.start_step - 1)
+            end = min(n_steps, subtask.end_step)
+            slice_steps = steps_list[start:end]
+            result = _generate_tips_for_segment(
+                task_description=subtask.generalized_description,
+                trajectory_slice="\n\n".join(slice_steps),
+                num_steps=len(slice_steps),
+                constrained_decoding_supported=constrained_decoding_supported,
+            )
+            results.append(result)
+        return results
+
+    # Fallback: full trajectory (use segmented description if exactly 1 subtask was found)
+    desc = subtasks[0].generalized_description if len(subtasks) == 1 else task_instruction
+    return [
+        _generate_tips_for_segment(
+            task_description=desc,
+            trajectory_slice=trajectory_data["trajectory_summary"],
+            num_steps=trajectory_data["num_steps"],
+            constrained_decoding_supported=constrained_decoding_supported,
+        )
+    ]

altk_evolve/schema/tips.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 from typing import Literal
 
 DEFAULT_TASK_DESCRIPTION = "Task description unknown"
@@ -17,6 +17,25 @@ class TipGenerationResponse(BaseModel):
     tips: list[Tip]
 
 
+class SubtaskSegment(BaseModel):
+    generalized_description: str = Field(
+        description="Value-agnostic description of the subtask, applicable to any agent performing a similar operation"
+    )
+    start_step: int = Field(ge=1, description="Inclusive 1-based start step index in the trajectory")
+    end_step: int = Field(ge=1, description="Inclusive 1-based end step index in the trajectory")
+    purpose: str = Field(description="What this subtask achieves (phase/output-oriented)")
+
+    @model_validator(mode="after")
+    def _check_range(self) -> "SubtaskSegment":
+        if self.end_step < self.start_step:
+            raise ValueError("end_step must be >= start_step")
+        return self
+
+
+class SegmentationResponse(BaseModel):
+    subtasks: list[SubtaskSegment] = Field(description="Contiguous, non-overlapping logical subtasks of the trajectory")
+
+
 @dataclass(frozen=True)
 class TipGenerationResult:
     """Internal result from generate_tips(), pairing tips with the source task description."""