Building On-Call Agents That Learn Using LangChain Memory

from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.language_models import BaseChatModel
import json
from pathlib import Path

class EpisodicMemoryStore:
    def __init__(self, store_path: str, llm: BaseChatModel):
        self.store_path = Path(store_path)
        self.store_path.mkdir(exist_ok=True)
        self.llm = llm

    def save_incident(self, incident: dict):
        """Save a resolved incident to episodic store."""
        path = self.store_path / f"{incident['incident_id']}.json"
        path.write_text(json.dumps(incident, indent=2))

    def recall_similar(self, service: str, error_hint: str, limit: int = 3) -> list[dict]:
        """Retrieve past incidents for the same service."""
        incidents = []
        for f in self.store_path.glob("*.json"):
            data = json.loads(f.read_text())
            if data.get("service") == service:
                incidents.append(data)

        # Sort by recency and cap results
        incidents.sort(
            key=lambda x: x.get("duration_minutes", 999)
        )
        return incidents[:limit]

    def summarize_for_prompt(self, incidents: list[dict]) -> str:
        """Compress past incidents into a prompt-friendly summary."""
        if not incidents:
            return "No past incidents found for this service."

        raw = json.dumps(incidents, indent=2)
        summary = self.llm.invoke(f"""
        Summarize these past incidents for an on-call agent.
        Focus on: what triggered each incident, what actions resolved it,
        and how long it took. Be concise - 3-4 bullet points max.

        Incidents:
        {raw}
        """)
        return summary.content

from langchain.memory import ConversationSummaryMemory

class SemanticMemoryStore:
    def __init__(self, llm: BaseChatModel):
        self.llm = llm
        self.patterns: dict[str, list[str]] = {}  # service -> learned patterns

    def extract_and_store(self, incident: dict):
        """Extract a reusable pattern from a resolved incident."""
        service = incident["service"]

        pattern_prompt = f"""
        An on-call agent just resolved this incident:
        Service: {service}
        Actions that worked: {incident.get("actions_taken", [])}
        Root cause: {incident.get("resolved_hypothesis", "unknown")}
        Duration: {incident.get("duration_minutes")} minutes

        Extract ONE reusable pattern for future incidents on this service.
        Format: "When [symptom], check [thing] because [reason]. Fix: [action]."
        One sentence only.
        """
        pattern = self.llm.invoke(pattern_prompt).content.strip()

        if service not in self.patterns:
            self.patterns[service] = []
        self.patterns[service].append(pattern)

        # Cap patterns per service to avoid bloat
        self.patterns[service] = self.patterns[service][-10:]

    def get_patterns(self, service: str) -> str:
        patterns = self.patterns.get(service, [])
        if not patterns:
            return "No learned patterns yet for this service."
        return "Learned patterns for this service:\n" + "\n".join(
            f"- {p}" for p in patterns
        )

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.language_models import BaseChatModel

class MemoryRouter:
    def __init__(
        self,
        episodic: EpisodicMemoryStore,
        semantic: SemanticMemoryStore,
        llm: BaseChatModel
    ):
        self.episodic = episodic
        self.semantic = semantic
        self.llm = llm

    def assemble_context(
        self,
        working: WorkingMemory,
        error_hint: str
    ) -> str:
        # Pull relevant episodic memories
        past_incidents = self.episodic.recall_similar(
            service=working.service,
            error_hint=error_hint,
            limit=3
        )
        episodic_summary = self.episodic.summarize_for_prompt(past_incidents)

        # Pull learned patterns
        semantic_patterns = self.semantic.get_patterns(working.service)

        # Assemble the full memory context for the agent prompt
        return f"""
        === MEMORY CONTEXT ===

        {working.to_prompt_context()}

        Past incidents on {working.service}:
        {episodic_summary}

        {semantic_patterns}

        === END MEMORY CONTEXT ===
        Use the above context to inform your investigation. 
        Prioritize patterns that match the current symptoms.
        Do not repeat actions that failed in past incidents unless you have new information.
        """

    def on_incident_close(self, working: WorkingMemory):
        """Called when incident resolves - persist to episodic and semantic stores."""
        record = working.close()
        self.episodic.save_incident(record)
        self.semantic.extract_and_store(record)

from langchain.agents import AgentExecutor, create_react_agent
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

# Initialize stores
episodic_store = EpisodicMemoryStore(store_path="./incident_memory", llm=llm)
semantic_store = SemanticMemoryStore(llm=llm)
router = MemoryRouter(episodic_store, semantic_store, llm)

def handle_incident(incident_id: str, service: str, alert: str):
    # Open working memory for this incident
    working = WorkingMemory(
        incident_id=incident_id,
        service=service,
        started_at=datetime.utcnow()
    )

    # Assemble memory context
    memory_context = router.assemble_context(working, error_hint=alert)

    # Build the agent prompt with memory injected
    full_prompt = f"""
    You are an on-call SRE agent. Use your memory context to investigate and resolve this incident.

    {memory_context}

    Current alert: {alert}

    Investigate and resolve this incident. If you recognize a pattern from past incidents, try that resolution first.
    """

    agent = create_react_agent(llm, tools, prompt_template)
    executor = AgentExecutor(
        agent=agent,
        tools=tools,
        max_iterations=10,
        max_execution_time=120,
        handle_parsing_errors=True
    )

    try:
        result = executor.invoke({"input": full_prompt})
        working.current_hypothesis = result["output"]
    finally:
        # Always persist - even if agent errored, save what we observed
        router.on_incident_close(working)

    return result