Extract learnings

Author: danielmillerp

examples/demos/procurement_agent/project/agents/extract_learnings_agent.py

@@ -0,0 +1,32 @@
+"""Agent for extracting learnings from human interactions."""
+
+from agents import Agent
+from datetime import timedelta
+
+
+def new_extract_learnings_agent() -> Agent:
+    """
+    Create an agent that extracts 1-2 sentence learnings from human interactions.
+
+    This agent analyzes the full conversation context to understand how we got to
+    the human interaction and what key insight or decision was made.
+
+    Returns:
+        Agent configured to extract a concise learning
+    """
+    instructions = """
+You are a learning extraction agent for a procurement system.
+
+Your job is to analyze only the wait_for_human tool call and extract a concise 1-2 sentence learning that can be applied to future decisions.
+
+The rest of the information is just context but the focus should be on understanding what the human wanted to do and why.
+
+Please extract a 1-2 sentence learning from the wait_for_human tool call.
+"""
+
+    return Agent(
+        name="Extract Learnings Agent",
+        instructions=instructions,
+        model="gpt-4o",
+        tools=[],  # No tools needed - just analysis
+    )

examples/demos/procurement_agent/project/scripts/send_test_events.py

@@ -41,60 +41,92 @@ async def send_fake_events(workflow_id: str):
     handle = client.get_workflow_handle(workflow_id)
 
     # Define the procurement event flow for Steel Beams
-    item = "Steel Beams"
-
-    fake_events = [
+    steel_beams_events = [
         SubmitalApprovalEvent(
             event_type=EventType.SUBMITTAL_APPROVED,
-            item=item
+            item="Steel Beams"
         ),
         ShipmentDepartedFactoryEvent(
             event_type=EventType.SHIPMENT_DEPARTED_FACTORY,
-            item=item,
+            item="Steel Beams",
             eta=datetime(2026, 2, 15)
         ),
         ShipmentArrivedSiteEvent(
             event_type=EventType.SHIPMENT_ARRIVED_SITE,
-            item=item,
+            item="Steel Beams",
             date_arrived=datetime(2026, 2, 15)
         ),
         InspectionFailedEvent(
             event_type=EventType.INSPECTION_FAILED,
-            item=item,
+            item="Steel Beams",
             inspection_date=datetime(2026, 2, 16)
         )
     ]
 
+    # Define the procurement event flow for HVAC Units
+    hvac_events = [
+        SubmitalApprovalEvent(
+            event_type=EventType.SUBMITTAL_APPROVED,
+            item="HVAC Units"
+        ),
+        ShipmentDepartedFactoryEvent(
+            event_type=EventType.SHIPMENT_DEPARTED_FACTORY,
+            item="HVAC Units",
+            eta=datetime(2026, 3, 10)
+        ),
+        ShipmentArrivedSiteEvent(
+            event_type=EventType.SHIPMENT_ARRIVED_SITE,
+            item="HVAC Units",
+            date_arrived=datetime(2026, 3, 10)
+        ),
+        InspectionFailedEvent(
+            event_type=EventType.INSPECTION_FAILED,
+            item="HVAC Units",
+            inspection_date=datetime(2026, 3, 11)
+        )
+    ]
+
+    # Combine all events
+    all_events = [
+        ("Steel Beams", steel_beams_events),
+        ("HVAC Units", hvac_events)
+    ]
+
     print(f"Connected to workflow: {workflow_id}")
     print("=" * 60)
-    print("Sending procurement events for Steel Beams...")
+    print("Sending procurement events...")
     print("=" * 60)
 
-    for i, event in enumerate(fake_events, 1):
-        print(f"\n[Event {i}] Sending: {event.event_type.value}")
-        print(f"  Item: {event.item}")
-
-        # Show additional details based on event type
-        if hasattr(event, 'eta'):
-            print(f"  ETA: {event.eta}")
-        if hasattr(event, 'date_arrived'):
-            print(f"  Date Arrived: {event.date_arrived}")
-        if hasattr(event, 'inspection_date'):
-            print(f"  Inspection Date: {event.inspection_date}")
-
-        try:
-            # Send the event using the send_event signal
-            # Convert event to JSON string
-            event_data = event.model_dump_json()
-            await handle.signal("send_event", event_data)
-            print(f"✓ Event sent successfully!")
-
-            # Wait a bit between events so you can see them being processed
-            await asyncio.sleep(2)
-
-        except Exception as e:
-            print(f"✗ Error sending event: {e}")
-            logger.error(f"Failed to send event: {e}")
+    for item_name, events in all_events:
+        print(f"\n{'=' * 60}")
+        print(f"Processing: {item_name}")
+        print("=" * 60)
+
+        for i, event in enumerate(events, 1):
+            print(f"\n[Event {i}] Sending: {event.event_type.value}")
+            print(f"  Item: {event.item}")
+
+            # Show additional details based on event type
+            if hasattr(event, 'eta'):
+                print(f"  ETA: {event.eta}")
+            if hasattr(event, 'date_arrived'):
+                print(f"  Date Arrived: {event.date_arrived}")
+            if hasattr(event, 'inspection_date'):
+                print(f"  Inspection Date: {event.inspection_date}")
+
+            try:
+                # Send the event using the send_event signal
+                # Convert event to JSON string
+                event_data = event.model_dump_json()
+                await handle.signal("send_event", event_data)
+                print(f"✓ Event sent successfully!")
+
+                # Wait a bit between events so you can see them being processed
+                await asyncio.sleep(2)
+
+            except Exception as e:
+                print(f"✗ Error sending event: {e}")
+                logger.error(f"Failed to send event: {e}")
 
     print("\n" + "=" * 60)
     print("All events have been sent!")

examples/demos/procurement_agent/project/utils/__init__.py

@@ -0,0 +1,5 @@
+"""Utility functions for the procurement agent."""
+
+from project.utils.learning_extraction import get_new_wait_for_human_context
+
+__all__ = ["get_new_wait_for_human_context"]

examples/demos/procurement_agent/project/utils/learning_extraction.py

@@ -0,0 +1,67 @@
+"""Utility for extracting new context from human interactions using a "going backwards" approach.
+
+This module prevents re-processing old wait_for_human calls by:
+1. Iterating backwards through the conversation
+2. Stopping when we hit a previously-processed wait_for_human call
+3. Returning only the NEW portion of the conversation
+"""
+
+from typing import List, Dict, Any, Set, Optional, Tuple
+from agentex.lib.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+
+def get_new_wait_for_human_context(
+    full_conversation: List[Dict[str, Any]],
+    extracted_learning_call_ids: Set[str],
+) -> Optional[Tuple[List[Dict[str, Any]], str]]:
+    """
+    Extract NEW context since the last processed wait_for_human call.
+
+    Similar to OpenCode's filterCompacted() pattern, this function:
+    - Iterates backwards through the full conversation history
+    - Stops when it finds a wait_for_human call we've already processed
+    - Returns only the NEW context
+
+    Args:
+        full_conversation: The complete conversation history (self._state.input_list)
+        extracted_learning_call_ids: Set of call_ids we've already extracted learnings from
+
+    Returns:
+        Tuple of (new_context_messages, call_id) if a new wait_for_human was found, None otherwise
+    """
+    # Go backwards through the conversation to find new wait_for_human calls
+    new_context = []
+    found_new_wait_for_human = False
+    new_wait_for_human_call_id = None
+
+    for item in reversed(full_conversation):
+        # Always collect items as we go backwards
+        new_context.append(item)
+
+        # Check if this is a wait_for_human function call
+        if isinstance(item, dict) and item.get("type") == "function_call":
+            if item.get("name") == "wait_for_human":
+                call_id = item.get("call_id")
+
+                # If we've already extracted learning for this call_id, STOP
+                if call_id in extracted_learning_call_ids:
+                    logger.info(f"Found already-processed wait_for_human call_id: {call_id}, stopping")
+                    break
+
+                # This is a NEW wait_for_human call
+                if not found_new_wait_for_human:
+                    found_new_wait_for_human = True
+                    new_wait_for_human_call_id = call_id
+                    logger.info(f"Found NEW wait_for_human call_id: {call_id}")
+
+    # If we found a new wait_for_human call, return the new context
+    if found_new_wait_for_human:
+        # Reverse back to chronological order
+        new_context.reverse()
+        logger.info(f"Extracted {len(new_context)} messages of new context")
+        return (new_context, new_wait_for_human_call_id)
+    else:
+        logger.info("No new wait_for_human calls found")
+        return None

examples/demos/procurement_agent/project/workflow.py

@@ -19,6 +19,8 @@
 from agentex.lib.core.temporal.plugins.openai_agents.hooks.hooks import TemporalStreamingHooks
 
 from project.agents.procurement_agent import new_procurement_agent
+from project.agents.extract_learnings_agent import new_extract_learnings_agent
+from project.utils.learning_extraction import get_new_wait_for_human_context
 from project.activities.activities import create_master_construction_schedule, get_master_construction_schedule
 from project.models.events import (
     EventType,
@@ -65,6 +67,7 @@ def __init__(self):
         self.human_queue: asyncio.Queue = asyncio.Queue()   # Human input
         self.event_log: list = []
         self.human_input_learnings: list = []
+        self.extracted_learning_call_ids: set = set()  # Track which wait_for_human calls we've extracted learnings from
 
         # Define activity retry policy with exponential backoff
         # Based on Temporal best practices from blog post
@@ -227,6 +230,31 @@ async def on_task_create(self, params: CreateTaskParams) -> str:
                     self._state.input_list = result.to_input_list()
                     logger.info(f"Successfully processed event at turn {self._state.turn_number}")
 
+                    # Extract learnings from NEW wait_for_human calls only (using going backwards approach)
+                    try:
+                        result_context = get_new_wait_for_human_context(
+                            full_conversation=self._state.input_list,
+                            extracted_learning_call_ids=self.extracted_learning_call_ids,
+                        )
+
+                        if result_context is not None:
+                            new_context, call_id = result_context
+                            logger.info("Found new wait_for_human call, extracting learning...")
+
+                            # Create extraction agent and run with only the NEW context
+                            extract_agent = new_extract_learnings_agent()
+                            extraction_result = await Runner.run(extract_agent, new_context, hooks=hooks)
+
+                            # Append the learning and track the call_id
+                            learning = extraction_result.final_output
+                            if learning:
+                                self.human_input_learnings.append(learning)
+                                self.extracted_learning_call_ids.add(call_id)
+                                logger.info(f"Extracted learning: {learning}")
+
+                    except Exception as e:
+                        logger.error(f"Failed to extract learning: {e}")
+
                 except Exception as e:
                     # Agent execution failed - graceful degradation
                     logger.error(f"Agent execution failed processing event: {e}")