feat: OCG-backed long-term agent memory with human good/bad feedback links

examples/agents/120_ocg_memory.py

@@ -0,0 +1,91 @@
+# Copyright (c) 2025 Agentspan
+# Licensed under the MIT License.
+
+"""120 — OCG-backed long-term memory with human good/bad feedback links.
+
+Enable memory on an agent and the runtime does two things automatically:
+
+  - BEFORE a run: relevant past memories (scoped to this agent/user) are
+    retrieved from OCG and injected into the prompt — no tool call needed.
+  - AFTER a run: the conversation is summarized (Claude-style: durable facts,
+    not the raw transcript) by a small internal summarizer agent and saved back
+    to OCG as a memory.
+
+Feedback is HUMAN-only. Agents never vote. Instead, the runtime hands a
+``FeedbackEvent`` — including signed *capability URLs* (good/bad) — to the
+agent's ``feedback_sink``. A human (e.g. a support engineer) clicks a link to
+mark the memory good or bad; the link skips auth (its signature is the
+authorization), so the clicker needs no OCG account. Here the sink just prints
+the URLs as they'd appear in a Zendesk ticket comment.
+
+Requires the OCG instance to be started with a feedback-link secret
+(``OCG_FEEDBACK_LINK_SECRET``) for the capability URLs to be minted.
+
+Run (from the repo root)::
+
+    OCG_INSTANCE_URL=https://test.contextgraph.io \
+    OCG_TOKEN=<bearer-token> \
+    uv run python examples/agents/120_ocg_memory.py
+
+    # against an embedded server, also set AGENTSPAN_SERVER_URL.
+"""
+
+import os
+
+from conductor.ai.agents import Agent, AgentRuntime, OCGMemoryStore, SemanticMemory
+from conductor.ai.agents.ocg_memory import FeedbackEvent
+
+MODEL = os.environ.get("AGENTSPAN_LLM_MODEL", "openai/gpt-4o-mini")
+
+OCG_INSTANCE_URL = os.environ.get("OCG_INSTANCE_URL") or ""
+# Unlike the ocg.py retrieval tools (which resolve a credential server-side),
+# the memory store calls OCG directly from Python, so it holds the bearer token.
+OCG_TOKEN = os.environ.get("OCG_TOKEN")
+if not OCG_INSTANCE_URL:
+    raise SystemExit("Set OCG_INSTANCE_URL to your OCG instance, e.g. https://test.contextgraph.io")
+
+
+def zendesk_sink(event: FeedbackEvent) -> None:
+    """Deliver the good/bad links to a human. In production this would POST a
+    comment to the Zendesk ticket; here we just print what would be sent."""
+    print("\n--- would post to Zendesk ticket ---")
+    print(f"Saved memory: {event.memory_key}")
+    print(f"Summary: {event.summary}")
+    if event.good_url:
+        print(f"  👍 Was this helpful?  {event.good_url}")
+        print(f"  👎 Not helpful:       {event.bad_url}")
+    print("------------------------------------\n")
+
+
+def main() -> None:
+    store = OCGMemoryStore(
+        url=OCG_INSTANCE_URL,
+        agent="agent:support",
+        user="user:alice",
+        token=OCG_TOKEN,
+    )
+
+    agent = Agent(
+        name="support",
+        model=MODEL,
+        instructions=(
+            "You are a customer support agent. Use any relevant context from "
+            "memory to personalize your answer. A memory labeled [bad] was "
+            "flagged by a human — treat it with suspicion."
+        ),
+        semantic_memory=SemanticMemory(store=store, max_results=5),
+        feedback_sink=zendesk_sink,
+    )
+
+    with AgentRuntime() as runtime:
+        print("--- Turn 1 ---")
+        runtime.run(
+            agent, "Hi, I'm Alice. I'm on the Enterprise plan and prefer email."
+        ).print_result()
+
+        print("\n--- Turn 2 (should recall Alice's plan from memory) ---")
+        runtime.run(agent, "What plan am I on again?").print_result()
+
+
+if __name__ == "__main__":
+    main()

src/conductor/ai/agents/__init__.py

@@ -193,6 +193,12 @@ def resolve_credentials(task: object, names: list) -> dict:
     schedules,
 )
 from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore, SemanticMemory
+from conductor.ai.agents.ocg_memory import (
+    FeedbackEvent,
+    MemorySummary,
+    OCGMemoryStore,
+    build_memory_summarizer,
+)
 
 # Termination conditions
 from conductor.ai.agents.termination import (
@@ -326,6 +332,10 @@ def resolve_credentials(task: object, names: list) -> dict:
     "SemanticMemory",
     "MemoryStore",
     "MemoryEntry",
+    "OCGMemoryStore",
+    "MemorySummary",
+    "FeedbackEvent",
+    "build_memory_summarizer",
     # Code execution
     "CodeExecutionConfig",
     "CliConfig",

src/conductor/ai/agents/agent.py

@@ -562,6 +562,9 @@ def __init__(
         output_type: Optional[type] = None,
         guardrails: Optional[List[Any]] = None,
         memory: Optional[Any] = None,
+        semantic_memory: Optional[Any] = None,
+        memory_summary_model: Optional[str] = None,
+        feedback_sink: Optional[Callable[..., Any]] = None,
         dependencies: Optional[Dict[str, Any]] = None,
         max_turns: int = 25,
         max_tokens: Optional[int] = None,
@@ -724,6 +727,13 @@ def __init__(
         self.output_type = output_type
         self.guardrails: List[Any] = list(guardrails) if guardrails else []
         self.memory = memory
+        # OCG-backed long-term memory (see agents/ocg_memory.py). When set, the
+        # runtime auto-injects relevant memories into the prompt before a run and,
+        # after the run, summarizes the conversation into a memory. feedback_sink,
+        # if provided, receives the good/bad capability links for that memory.
+        self.semantic_memory = semantic_memory
+        self.memory_summary_model = memory_summary_model
+        self.feedback_sink = feedback_sink
         self.dependencies: Dict[str, Any] = dict(dependencies) if dependencies else {}
         self.max_turns = max_turns
         self.max_tokens = max_tokens

src/conductor/ai/agents/config_serializer.py

@@ -130,10 +130,21 @@ def _serialize_agent(self, agent: "Agent") -> dict:
         if agent.guardrails:
             config["guardrails"] = [self._serialize_guardrail(g) for g in agent.guardrails]
 
-        # Memory
+        # Memory (short-term conversation)
         if hasattr(agent, "memory") and agent.memory:
             config["memory"] = self._serialize_memory(agent.memory)
 
+        # Long-term (OCG-backed) memory. When present, the server-side compiler
+        # inlines retrieval (pre-loop) + distill/save/feedback (post-loop) steps
+        # so memory works on the deployed/webhook path — not just client run().
+        ltm = self._serialize_long_term_memory(agent)
+        if ltm is not None:
+            config["longTermMemory"] = ltm
+            # feedback_sink delivers the human good/bad capability links out-of-band.
+            # Emit a worker ref so the compiled path can call the Python sink worker.
+            if getattr(agent, "feedback_sink", None) is not None:
+                config["feedbackSink"] = {"taskName": f"{agent.name}_feedback_sink"}
+
         # Max tokens
         if agent.max_tokens is not None:
             config["maxTokens"] = agent.max_tokens
@@ -508,3 +519,34 @@ def _serialize_memory(self, memory: Any) -> dict:
         if hasattr(memory, "max_messages") and memory.max_messages:
             result["maxMessages"] = memory.max_messages
         return result
+
+    def _serialize_long_term_memory(self, agent: "Agent") -> "Any":
+        """Serialize an agent's OCG-backed semantic memory to a LongTermMemoryConfig dict.
+
+        Returns ``None`` (no-op) unless the agent has a ``semantic_memory`` whose
+        store exposes an OCG base url. Reads the OCG instance url, scope owner,
+        user and scope off the store; the credential is a SERVER-resolvable secret
+        NAME (e.g. ``OCG_PUBLIC_KEY``) — never the raw client token. The summary
+        model falls back to the agent's own model when not explicitly set.
+        """
+        sm = getattr(agent, "semantic_memory", None)
+        if sm is None:
+            return None
+        store = getattr(sm, "store", None)
+        # Only OCG-backed stores compile server-side (need a base url to call).
+        base = getattr(store, "_base", None) if store is not None else None
+        if not base:
+            return None
+
+        result: Dict[str, Any] = {
+            "ocgUrl": base,
+            "credential": getattr(store, "_credential", None) or "OCG_PUBLIC_KEY",
+            "agent": getattr(store, "_agent", None),
+            "scope": getattr(store, "_scope", None) or "agent",
+            "maxResults": getattr(sm, "max_results", None),
+            "summaryModel": getattr(agent, "memory_summary_model", None) or (agent.model or None),
+        }
+        user = getattr(store, "_user", None)
+        if user:
+            result["user"] = user
+        return {k: v for k, v in result.items() if v is not None}