diff --git a/sdk/python/examples/118_ocg_memory.py b/sdk/python/examples/118_ocg_memory.py new file mode 100644 index 00000000..da982d7f --- /dev/null +++ b/sdk/python/examples/118_ocg_memory.py @@ -0,0 +1,91 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. + +"""118 — 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 ``sdk/python``):: + + OCG_INSTANCE_URL=https://test.contextgraph.io \ + OCG_TOKEN= \ + uv run python examples/118_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() diff --git a/sdk/python/src/conductor/ai/agents/__init__.py b/sdk/python/src/conductor/ai/agents/__init__.py index 16a4e163..243dca52 100644 --- a/sdk/python/src/conductor/ai/agents/__init__.py +++ b/sdk/python/src/conductor/ai/agents/__init__.py @@ -195,6 +195,12 @@ def resolve_credentials(input_data: dict, 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 ( @@ -327,6 +333,10 @@ def resolve_credentials(input_data: dict, names: list) -> dict: "SemanticMemory", "MemoryStore", "MemoryEntry", + "OCGMemoryStore", + "MemorySummary", + "FeedbackEvent", + "build_memory_summarizer", # Code execution "CodeExecutionConfig", "CliConfig", diff --git a/sdk/python/src/conductor/ai/agents/agent.py b/sdk/python/src/conductor/ai/agents/agent.py index f79f9dca..6b886ccf 100644 --- a/sdk/python/src/conductor/ai/agents/agent.py +++ b/sdk/python/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 diff --git a/sdk/python/src/conductor/ai/agents/config_serializer.py b/sdk/python/src/conductor/ai/agents/config_serializer.py index ee68c0ba..d153e531 100644 --- a/sdk/python/src/conductor/ai/agents/config_serializer.py +++ b/sdk/python/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} diff --git a/sdk/python/src/conductor/ai/agents/ocg_memory.py b/sdk/python/src/conductor/ai/agents/ocg_memory.py new file mode 100644 index 00000000..69f5500b --- /dev/null +++ b/sdk/python/src/conductor/ai/agents/ocg_memory.py @@ -0,0 +1,284 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. See LICENSE file in the project root for details. + +"""OCG-backed long-term memory for agents. + +This module backs agentspan's :class:`~conductor.ai.agents.semantic_memory.MemoryStore` +abstraction with an OCG (Open Context Graph) instance, so an agent's memories +persist in OCG and ride OCG's feedback-aware ranking. + +Three pieces: + +- :class:`OCGMemoryStore` — a synchronous HTTP adapter implementing ``MemoryStore`` + (``add`` / ``search`` / ``delete`` / ``clear`` / ``list_all``) against the OCG BFF. +- :class:`MemorySummary` + :func:`build_memory_summarizer` — a small agent that + distills a conversation into durable facts (used by the runtime's post-run save). +- :class:`FeedbackEvent` — what the runtime hands to an Agent's ``feedback_sink`` + after saving a memory: the distilled summary plus signed *capability URLs* a human + can click to mark the memory good/bad (no OCG account needed). + +Design notes: + +- The OCG bearer ``token`` is held **client-side** here (e.g. from ``OCG_TOKEN``), + unlike the ``ocg.py`` retrieval tools which resolve a credential server-side. +- Agents only ever **create and read** memories. Good/bad feedback is human-only: + it is delivered out-of-band through ``feedback_sink`` (e.g. into a Zendesk ticket) + and the capability URLs are never surfaced to the agent's LLM. +""" + +from __future__ import annotations + +import hashlib +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +import httpx + +from conductor.ai.agents.exceptions import AgentAPIError +from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore + +if TYPE_CHECKING: + from conductor.ai.agents.agent import Agent + +logger = logging.getLogger("conductor.ai.agents.ocg_memory") + + +def _hash_key(content: str) -> str: + return "mem-" + hashlib.sha256(content.encode("utf-8")).hexdigest()[:16] + + +class OCGMemoryStore(MemoryStore): + """Back agentspan :class:`SemanticMemory` with an OCG instance. + + Implements the synchronous ``MemoryStore`` interface over the OCG BFF: + + - ``add`` -> ``POST /api/v1/memories`` + - ``search`` -> ``POST /api/v1/memories/search`` (feedback-blended ranking) + - ``delete`` -> ``DELETE /api/v1/memories/{key}`` + - ``list_all``-> ``GET /api/v1/memories`` + + Args: + url: Base URL of the OCG instance (required). + agent: Agent owner key, e.g. ``"agent:support"`` (required). + user: Optional user owner, e.g. ``"user:alice"``. + token: OCG bearer token, held client-side (e.g. from ``OCG_TOKEN``). + Used by the client-side ``run()`` path. + credential: Server-resolvable credential NAME (default ``"OCG_PUBLIC_KEY"``) + for the OCG bearer token. Used by the COMPILED/deployed path — the + server resolves this via a ``#{NAME}`` HTTP-header placeholder. Distinct + from ``token`` (the raw client token); both can coexist. + scope: Memory scope for writes (default ``"user"``). + timeout: Per-request timeout in seconds. + client: Optional pre-built ``httpx.Client`` (mainly for tests). + """ + + def __init__( + self, + *, + url: str, + agent: str, + user: Optional[str] = None, + token: Optional[str] = None, + credential: str = "OCG_PUBLIC_KEY", + scope: str = "user", + timeout: float = 10.0, + client: Optional[httpx.Client] = None, + ) -> None: + if not url or not url.strip(): + raise ValueError("OCGMemoryStore requires a non-blank OCG instance url") + if not agent or not agent.strip(): + raise ValueError("OCGMemoryStore requires a non-blank agent owner") + self._base = url.strip().rstrip("/") + self._agent = agent + self._user = user + self._credential = credential + self._scope = scope + headers: Dict[str, str] = {} + if token: + headers["Authorization"] = f"Bearer {token}" + self._client = client or httpx.Client(timeout=timeout, headers=headers) + + # ── HTTP plumbing ─────────────────────────────────────────────────── + + def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: + try: + resp = self._client.request(method, self._base + path, **kwargs) + except httpx.HTTPError as exc: # network/timeout + raise AgentAPIError(status_code=0, message=str(exc), url=self._base + path) from exc + if resp.status_code >= 400: + raise AgentAPIError( + status_code=resp.status_code, + message=resp.text, + url=self._base + path, + ) + return resp + + # ── MemoryStore interface ─────────────────────────────────────────── + + def add(self, entry: MemoryEntry) -> str: + key = entry.id or str(entry.metadata.get("key") or "") or _hash_key(entry.content) + body: Dict[str, Any] = { + "key": key, + "agent": self._agent, + "value": entry.content, + "description": entry.content[:200], + "scope": self._scope, + "source": "agent_inferred", + "tags": list(entry.metadata.get("tags", []) or []), + } + if self._user: + body["user"] = self._user + self._request("POST", "/api/v1/memories", json=body) + entry.id = key + return key + + def search(self, query: str, top_k: int = 5) -> List[MemoryEntry]: + body: Dict[str, Any] = { + "query": query, + "agent": self._agent, + "limit": top_k, + "include_shared": True, + } + if self._user: + body["user"] = self._user + resp = self._request("POST", "/api/v1/memories/search", json=body) + out: List[MemoryEntry] = [] + for m in resp.json().get("memories", []) or []: + out.append( + MemoryEntry( + id=m.get("key", ""), + content=_with_signal(m.get("value_preview", ""), m), + metadata={ + "relevance_score": m.get("relevance_score"), + "good_count": m.get("good_count", 0), + "bad_count": m.get("bad_count", 0), + }, + ) + ) + return out + + def delete(self, memory_id: str) -> bool: + params: Dict[str, str] = {"agent": self._agent} + if self._user: + params["user"] = self._user + try: + self._request("DELETE", f"/api/v1/memories/{memory_id}", params=params) + except AgentAPIError: + return False + return True + + def clear(self) -> None: + # No bulk-clear endpoint — fan out over the listed keys. Guard usage: + # this deletes every memory for the configured agent/user. + entries = self.list_all() + logger.warning( + "OCGMemoryStore.clear() deleting %d memories for %s", len(entries), self._agent + ) + for e in entries: + self.delete(e.id) + + def list_all(self) -> List[MemoryEntry]: + params: Dict[str, str] = {"agent": self._agent, "limit": "200"} + if self._user: + params["user"] = self._user + resp = self._request("GET", "/api/v1/memories", params=params) + return [ + MemoryEntry(id=m.get("key", ""), content=m.get("value_preview", "")) + for m in resp.json().get("memories", []) or [] + ] + + # ── Capability feedback links (human-only, out-of-band) ───────────── + + def feedback_links(self, key: str) -> Dict[str, Any]: + """Mint signed good/bad capability URLs for a memory. + + Returns ``{"good_url", "bad_url", "expires_at"}``. The URLs require no OCG + login — a human (e.g. a support engineer) clicks them to vote. Requires the + OCG instance to have a feedback-link secret configured (else OCG returns 501). + """ + params: Dict[str, str] = {"agent": self._agent} + if self._user: + params["user"] = self._user + resp = self._request("POST", f"/api/v1/memories/{key}/feedback-links", params=params) + return resp.json() + + +def _with_signal(content: str, m: Dict[str, Any]) -> str: + """Fold the human good/bad signal into a search result's content so the + injected prompt context shows the agent when a memory was marked bad and why.""" + good = int(m.get("good_count", 0) or 0) + bad = int(m.get("bad_count", 0) or 0) + if not good and not bad: + return content + content += f" [good {good} / bad {bad}]" + for note in m.get("feedback_notes") or []: + if note.get("verdict") == "bad" and note.get("reason"): + content += f' (bad: "{note["reason"]}")' + return content + + +# ── Conversation summarization (Claude-style distillation) ────────────── + +try: # pydantic is a core dep, but keep the import resilient. + from pydantic import BaseModel, Field + + class MemorySummary(BaseModel): + """Structured output for the conversation summarizer agent.""" + + summary: str = Field(description="One short paragraph: what happened / what was learned.") + facts: List[str] = Field( + default_factory=list, + description="Durable, reusable facts about the user or task (no chit-chat).", + ) + tags: List[str] = Field(default_factory=list, description="Short topical tags.") + +except Exception: # pragma: no cover - pydantic always present in practice + MemorySummary = None # type: ignore[assignment] + + +MEMORY_SUMMARIZER_INSTRUCTIONS = ( + "You distill a conversation into a durable memory. Read the transcript and " + "extract only reusable, durable facts about the user, their preferences, and " + "the task — the kind of thing worth remembering for next time. Ignore greetings, " + "filler, and one-off details. Write a one-paragraph summary, a short list of " + "facts, and a few topical tags. Be concise and concrete." +) + + +def build_memory_summarizer(model: str, *, name: str = "__memory_summarizer") -> "Agent": + """Build the internal agent that summarizes a conversation into a memory. + + It uses :class:`MemorySummary` structured output and is intentionally created + WITHOUT ``semantic_memory`` so the post-run save hook skips it (no recursion). + """ + from conductor.ai.agents.agent import Agent + + return Agent( + name=name, + model=model, + instructions=MEMORY_SUMMARIZER_INSTRUCTIONS, + output_type=MemorySummary, + max_turns=1, + ) + + +@dataclass +class FeedbackEvent: + """Handed to an Agent's ``feedback_sink`` after a conversation memory is saved. + + Carries the distilled summary plus the signed capability URLs a human can click + to mark the memory good/bad. The integrator routes these out-of-band (e.g. posts + them into a Zendesk ticket). These URLs are never shown to the agent's LLM. + """ + + memory_key: str + summary: str + facts: List[str] = field(default_factory=list) + tags: List[str] = field(default_factory=list) + good_url: Optional[str] = None + bad_url: Optional[str] = None + expires_at: Optional[str] = None + agent: Optional[str] = None + user: Optional[str] = None + session_id: Optional[str] = None diff --git a/sdk/python/src/conductor/ai/agents/runtime/runtime.py b/sdk/python/src/conductor/ai/agents/runtime/runtime.py index 486d2bbd..9e022e15 100644 --- a/sdk/python/src/conductor/ai/agents/runtime/runtime.py +++ b/sdk/python/src/conductor/ai/agents/runtime/runtime.py @@ -133,6 +133,38 @@ def _has_stateful_tools(agent: Any) -> bool: return False +def _agent_model_str(agent: Any) -> str: + """Best-effort string model id for the conversation summarizer.""" + m = getattr(agent, "model", "") + if isinstance(m, str) and m: + return m + return "openai/gpt-4o-mini" + + +def _parse_summary_output(output: Any) -> "tuple[str, list, list]": + """Extract (summary, facts, tags) from a summarizer run's output. + + Handles a pydantic MemorySummary, a plain dict (optionally wrapped under a + ``result`` key), or any other value (stringified as the summary). + """ + # pydantic model instance + if hasattr(output, "summary"): + return ( + str(getattr(output, "summary", "") or ""), + list(getattr(output, "facts", []) or []), + list(getattr(output, "tags", []) or []), + ) + if isinstance(output, dict): + data = output.get("result", output) + if isinstance(data, dict): + return ( + str(data.get("summary", "") or ""), + list(data.get("facts", []) or []), + list(data.get("tags", []) or []), + ) + return (str(output) if output is not None else "", [], []) + + # Thread count for system-level async workers (guardrails, handoff checks, etc.). # User-defined tool workers keep the per-worker default from @worker_task. _SYSTEM_WORKER_THREADS = 10 @@ -1035,6 +1067,13 @@ def _collect_worker_names(self, agent: Agent, *, required_workers: Optional[set] if agent.termination: names.add(f"{agent.name}_termination") + # Long-term (OCG) memory feedback_sink — compiled path emits a SIMPLE + # task that delivers the human good/bad capability links out-of-band. + if getattr(agent, "semantic_memory", None) is not None and callable( + getattr(agent, "feedback_sink", None) + ): + names.add(f"{agent.name}_feedback_sink") + # Callable gate (sequential pipeline) if getattr(agent, "gate", None) is not None and callable(agent.gate): names.add(f"{agent.name}_gate") @@ -1190,6 +1229,16 @@ def _server_needs(task_name: str) -> bool: if _server_needs(task_name): self._register_stop_when_worker(agent.name, agent.stop_when, domain=domain) + # 3a. Long-term memory feedback_sink — compiled path hands the human + # good/bad capability links to this worker after a conversation memory + # is saved (mirrors run()'s post-run FeedbackEvent delivery). + if getattr(agent, "semantic_memory", None) is not None and callable( + getattr(agent, "feedback_sink", None) + ): + task_name = f"{agent.name}_feedback_sink" + if _server_needs(task_name): + self._register_feedback_sink_worker(agent.name, agent.feedback_sink, domain=domain) + # 3b. Callbacks (legacy + CallbackHandler chaining) from conductor.ai.agents.callback import ( _LEGACY_ATTR_TO_POSITION, @@ -1569,6 +1618,75 @@ async def stop_when_worker(result="", iteration: int = 0, messages=None) -> obje lease_extend_enabled=True, )(stop_when_worker) + def _register_feedback_sink_worker( + self, agent_name: str, feedback_sink_fn, domain: "Optional[str]" = None + ) -> None: + """Register a long-term-memory feedback_sink worker. + + The compiled (server-side) memory path emits a SIMPLE task that, after + saving a conversation memory and minting the signed good/bad capability + URLs, invokes this worker with the FeedbackEvent fields. The worker + rebuilds a :class:`FeedbackEvent` and hands it to the user's + ``feedback_sink`` callable — mirroring run()'s out-of-band delivery. + Best-effort: failures are swallowed so memory never fails the run. + """ + from conductor.client.worker.worker_task import worker_task + + task_name = f"{agent_name}_feedback_sink" + + async def feedback_sink_worker( + memory_key: str = "", + summary: str = "", + facts: object = None, + tags: object = None, + good_url: str = None, + bad_url: str = None, + expires_at: str = None, + agent: str = None, + user: str = None, + ) -> object: + try: + from conductor.ai.agents.ocg_memory import FeedbackEvent + + event = FeedbackEvent( + memory_key=memory_key, + summary=summary, + facts=list(facts) if isinstance(facts, (list, tuple)) else [], + tags=list(tags) if isinstance(tags, (list, tuple)) else [], + good_url=good_url, + bad_url=bad_url, + expires_at=expires_at, + agent=agent, + user=user, + ) + await _call_user_fn(feedback_sink_fn, event) + return {"delivered": True} + except Exception as e: + logger.warning("feedback_sink delivery failed: %s", e) + return {"delivered": False} + + feedback_sink_worker.__annotations__ = { + "memory_key": str, + "summary": str, + "facts": object, + "tags": object, + "good_url": str, + "bad_url": str, + "expires_at": str, + "agent": str, + "user": str, + "return": object, + } + worker_task( + task_definition_name=task_name, + task_def=_default_task_def(task_name), + register_task_def=True, + overwrite_task_def=True, + domain=domain, + thread_count=_SYSTEM_WORKER_THREADS, + lease_extend_enabled=True, + )(feedback_sink_worker) + def _register_gate_worker( self, agent_name: str, gate_fn, domain: "Optional[str]" = None ) -> None: @@ -2725,6 +2843,9 @@ def run( resolved_prompt = self._check_input_guardrails(agent, resolved_prompt) self._validate_execution_input(resolved_prompt, media=media, context=context) + # OCG long-term memory: inject relevant past memories into the prompt. + agent = self._apply_memory_retrieval(agent, resolved_prompt) + correlation_id = str(uuid.uuid4()) logger.info("Executing agent '%s'", agent.name) @@ -2804,7 +2925,7 @@ def run( error_reason = task_failure_reason or status.reason logger.info("Agent '%s' completed (execution_id=%s)", agent.name, execution_id) - return AgentResult( + result = AgentResult( output=output, execution_id=execution_id, correlation_id=correlation_id, @@ -2817,6 +2938,118 @@ def run( sub_results=self._extract_sub_results(output), ) + # OCG long-term memory: summarize the conversation into a memory and + # hand the human good/bad capability links to the agent's feedback_sink. + self._maybe_save_conversation_memory(agent, result, session_id) + + return result + + # ── OCG long-term memory hooks ───────────────────────────────── + + def _apply_memory_retrieval(self, agent: "Agent", prompt: str) -> "Agent": + """Prepend relevant OCG memories to a COPY of the agent's instructions. + + No-op (returns the original agent) when the agent has no ``semantic_memory`` + or nothing relevant is found. Never mutates the shared agent instance. + """ + sm = getattr(agent, "semantic_memory", None) + if sm is None: + return agent + try: + context = sm.get_context(prompt) + except Exception as exc: # retrieval is best-effort + logger.warning("memory retrieval failed for agent '%s': %s", agent.name, exc) + return agent + if not context: + return agent + + import copy as _copy + + augmented = _copy.copy(agent) + base = agent.instructions if isinstance(agent.instructions, str) else "" + augmented.instructions = (context + "\n\n" + base) if base else context + return augmented + + def _maybe_save_conversation_memory( + self, agent: "Agent", result: "AgentResult", session_id: Optional[str] + ) -> None: + """Summarize the run's conversation into an OCG memory (best-effort). + + Runs an internal summarizer agent (server-side) to distill durable facts, + stores the result via the agent's ``semantic_memory`` store, then mints the + good/bad capability links and passes them to ``agent.feedback_sink``. All + failures are swallowed — memory saving must never fail the primary run. + """ + sm = getattr(agent, "semantic_memory", None) + if sm is None or getattr(result, "status", None) != "COMPLETED": + return + if not getattr(result, "messages", None): + return + try: + from conductor.ai.agents.ocg_memory import ( + FeedbackEvent, + build_memory_summarizer, + ) + + transcript = self._transcript_text(result.messages) + model = getattr(agent, "memory_summary_model", None) or _agent_model_str(agent) + summarizer = build_memory_summarizer(model) + summary_result = self.run(summarizer, transcript) + + summary, facts, tags = _parse_summary_output(summary_result.output) + content = summary + if facts: + content += "\n\nFacts:\n" + "\n".join(f"- {f}" for f in facts) + + key = f"conversation:{session_id or result.execution_id}" + store = sm.store + from conductor.ai.agents.semantic_memory import MemoryEntry + + store.add( + MemoryEntry( + id=key, + content=content, + metadata={"key": key, "tags": list(tags) + ["conversation"]}, + ) + ) + logger.info("saved conversation memory '%s' for agent '%s'", key, agent.name) + + sink = getattr(agent, "feedback_sink", None) + if sink is None: + return + links: Dict[str, Any] = {} + if hasattr(store, "feedback_links"): + try: + links = store.feedback_links(key) + except Exception as exc: + logger.warning("could not mint feedback links for '%s': %s", key, exc) + event = FeedbackEvent( + memory_key=key, + summary=summary, + facts=list(facts), + tags=list(tags), + good_url=links.get("good_url"), + bad_url=links.get("bad_url"), + expires_at=links.get("expires_at"), + agent=getattr(store, "_agent", None), + user=getattr(store, "_user", None), + session_id=session_id, + ) + sink(event) + except Exception as exc: # never fail the primary run + logger.warning("conversation memory save failed for agent '%s': %s", agent.name, exc) + + @staticmethod + def _transcript_text(messages: List[Dict[str, Any]]) -> str: + lines: List[str] = [] + for m in messages: + role = str(m.get("role", "")) or "?" + content = m.get("content", "") + if not isinstance(content, str): + content = json.dumps(content) + lines.append(f"{role}: {content}") + return "\n".join(lines) + # ── Run-by-name (pre-deployed agents) ────────────────────────── def _run_by_name( diff --git a/sdk/python/tests/unit/test_config_serializer.py b/sdk/python/tests/unit/test_config_serializer.py index 404427f0..d2c9bab7 100644 --- a/sdk/python/tests/unit/test_config_serializer.py +++ b/sdk/python/tests/unit/test_config_serializer.py @@ -214,6 +214,69 @@ def test_serialize_memory(self): assert config["memory"]["messages"] == [{"role": "system", "message": "context"}] + def test_serialize_long_term_memory(self): + """OCG-backed semantic_memory serializes to longTermMemory + feedbackSink.""" + from conductor.ai.agents.agent import Agent + from conductor.ai.agents.ocg_memory import OCGMemoryStore + from conductor.ai.agents.semantic_memory import SemanticMemory + + store = OCGMemoryStore( + url="https://ocg.example.com/", + agent="agent:ce-ticket-resolution", + user="user:alice", + scope="agent", + ) + sm = SemanticMemory(store=store, max_results=7) + + def sink(event): # feedback_sink callable + return None + + agent = Agent( + name="ce_agent", + model="openai/gpt-4o", + instructions="Resolve tickets.", + semantic_memory=sm, + memory_summary_model="openai/gpt-4o-mini", + feedback_sink=sink, + ) + config = self.serializer.serialize(agent) + + ltm = config["longTermMemory"] + assert ltm["ocgUrl"] == "https://ocg.example.com" # trailing slash stripped + assert ltm["credential"] == "OCG_PUBLIC_KEY" # server-resolvable name, not token + assert ltm["agent"] == "agent:ce-ticket-resolution" + assert ltm["user"] == "user:alice" + assert ltm["scope"] == "agent" + assert ltm["maxResults"] == 7 + assert ltm["summaryModel"] == "openai/gpt-4o-mini" + + assert config["feedbackSink"] == {"taskName": "ce_agent_feedback_sink"} + + def test_serialize_long_term_memory_absent(self): + """No semantic_memory -> no longTermMemory / feedbackSink keys (no-op).""" + from conductor.ai.agents.agent import Agent + + agent = Agent(name="plain", model="openai/gpt-4o", instructions="Hi.") + config = self.serializer.serialize(agent) + + assert "longTermMemory" not in config + assert "feedbackSink" not in config + + def test_serialize_long_term_memory_summary_model_fallback(self): + """summaryModel falls back to the agent's own model when unset.""" + from conductor.ai.agents.agent import Agent + from conductor.ai.agents.ocg_memory import OCGMemoryStore + from conductor.ai.agents.semantic_memory import SemanticMemory + + store = OCGMemoryStore(url="https://ocg.example.com", agent="agent:x") + sm = SemanticMemory(store=store, max_results=5) + agent = Agent(name="a", model="anthropic/claude", semantic_memory=sm) + config = self.serializer.serialize(agent) + + assert config["longTermMemory"]["summaryModel"] == "anthropic/claude" + # No feedback_sink -> no feedbackSink emitted. + assert "feedbackSink" not in config + def test_serialize_gate_text(self): """TextGate serializes to text_contains config.""" from conductor.ai.agents.agent import Agent diff --git a/sdk/python/tests/unit/test_ocg_memory_store.py b/sdk/python/tests/unit/test_ocg_memory_store.py new file mode 100644 index 00000000..d5710991 --- /dev/null +++ b/sdk/python/tests/unit/test_ocg_memory_store.py @@ -0,0 +1,253 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. See LICENSE file in the project root for details. + +"""Unit tests for OCG-backed memory: the HTTP store adapter, the conversation +summary helpers, and the runtime save/retrieval hooks.""" + +from __future__ import annotations + +import json +from typing import List + +import httpx +import pytest + +from conductor.ai.agents import Agent +from conductor.ai.agents.exceptions import AgentAPIError +from conductor.ai.agents.ocg_memory import FeedbackEvent, OCGMemoryStore +from conductor.ai.agents.result import AgentResult, Status +from conductor.ai.agents.runtime.runtime import ( + AgentRuntime, + _agent_model_str, + _parse_summary_output, +) +from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore, SemanticMemory + + +def _store_with(handler) -> OCGMemoryStore: + client = httpx.Client(transport=httpx.MockTransport(handler)) + return OCGMemoryStore(url="https://ocg.test", agent="agent:a", user="user:bob", client=client) + + +class TestOCGMemoryStore: + def test_add_posts_value_field_and_no_confidence(self): + captured = {} + + def handler(req: httpx.Request) -> httpx.Response: + captured["url"] = str(req.url) + captured["body"] = json.loads(req.content) + return httpx.Response(200, json={"key": "k1"}) + + store = _store_with(handler) + key = store.add(MemoryEntry(content="alice prefers email", metadata={"key": "pref"})) + + assert key == "pref" + assert captured["url"].endswith("/api/v1/memories") + body = captured["body"] + assert body["value"] == "alice prefers email" # field is "value", NOT "string_value" + assert "string_value" not in body + assert "confidence" not in body # confidence was removed from the API + assert body["agent"] == "agent:a" and body["user"] == "user:bob" + + def test_search_folds_good_bad_signal_into_content(self): + def handler(req: httpx.Request) -> httpx.Response: + assert str(req.url).endswith("/api/v1/memories/search") + return httpx.Response( + 200, + json={ + "memories": [ + { + "key": "m1", + "value_preview": "use us-east-1", + "good_count": 2, + "bad_count": 1, + "relevance_score": 0.9, + "feedback_notes": [{"verdict": "bad", "reason": "stale region"}], + } + ] + }, + ) + + store = _store_with(handler) + entries = store.search("which region", top_k=5) + assert len(entries) == 1 + assert "[good 2 / bad 1]" in entries[0].content + assert 'bad: "stale region"' in entries[0].content + + def test_feedback_links_hits_mint_route(self): + def handler(req: httpx.Request) -> httpx.Response: + assert str(req.url).split("?")[0].endswith("/api/v1/memories/k1/feedback-links") + return httpx.Response( + 200, + json={ + "good_url": "https://ocg.test/api/v1/feedback/GOOD", + "bad_url": "https://ocg.test/api/v1/feedback/BAD", + "expires_at": "2026-09-01T00:00:00Z", + }, + ) + + store = _store_with(handler) + links = store.feedback_links("k1") + assert links["good_url"].endswith("/feedback/GOOD") + assert links["bad_url"].endswith("/feedback/BAD") + + def test_non_2xx_raises(self): + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response(500, text="boom") + + store = _store_with(handler) + with pytest.raises(AgentAPIError): + store.add(MemoryEntry(content="x", metadata={"key": "k"})) + + +class _FakeStore(MemoryStore): + """Records add() calls and serves canned feedback links.""" + + def __init__(self): + self.added: List[MemoryEntry] = [] + self._agent = "agent:a" + self._user = "user:bob" + + def add(self, entry: MemoryEntry) -> str: + self.added.append(entry) + return entry.id or "k" + + def search(self, query: str, top_k: int = 5) -> List[MemoryEntry]: + return [] + + def delete(self, memory_id: str) -> bool: + return True + + def clear(self) -> None: + pass + + def list_all(self) -> List[MemoryEntry]: + return [] + + def feedback_links(self, key: str): + return { + "good_url": "https://ocg.test/api/v1/feedback/GOOD", + "bad_url": "https://ocg.test/api/v1/feedback/BAD", + "expires_at": "2026-09-01T00:00:00Z", + } + + +class TestRuntimeMemoryHooks: + def test_save_stores_distilled_summary_and_invokes_sink(self, monkeypatch): + rt = AgentRuntime() + store = _FakeStore() + events: List[FeedbackEvent] = [] + + agent = Agent( + name="support", + model="openai/gpt-4o", + semantic_memory=SemanticMemory(store=store), + feedback_sink=lambda ev: events.append(ev), + ) + + # Stub the nested summarizer run — return distilled facts, NOT the transcript. + def fake_run(summarizer_agent, transcript, **kwargs): + assert summarizer_agent.name == "__memory_summarizer" + return AgentResult( + output={ + "summary": "Alice is on Enterprise.", + "facts": ["plan=enterprise"], + "tags": ["billing"], + }, + status=Status.COMPLETED, + ) + + monkeypatch.setattr(rt, "run", fake_run) + + result = AgentResult( + output={"result": "ok"}, + execution_id="exec-1", + status=Status.COMPLETED, + messages=[{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}], + ) + + rt._maybe_save_conversation_memory(agent, result, session_id="sess-9") + + assert len(store.added) == 1 + saved = store.added[0] + assert saved.id == "conversation:sess-9" + assert "Alice is on Enterprise." in saved.content + assert "plan=enterprise" in saved.content + assert "hello" not in saved.content # the raw transcript is NOT stored + assert "conversation" in saved.metadata["tags"] + + assert len(events) == 1 + assert events[0].good_url.endswith("/feedback/GOOD") + assert events[0].bad_url.endswith("/feedback/BAD") + assert events[0].memory_key == "conversation:sess-9" + + def test_save_skipped_when_no_semantic_memory(self, monkeypatch): + rt = AgentRuntime() + called = {"run": False} + monkeypatch.setattr(rt, "run", lambda *a, **k: called.__setitem__("run", True)) + agent = Agent(name="plain", model="openai/gpt-4o") + rt._maybe_save_conversation_memory( + agent, + AgentResult(status=Status.COMPLETED, messages=[{"role": "user", "content": "hi"}]), + None, + ) + assert called["run"] is False # no nested summarizer run, no recursion + + def test_save_never_raises_on_failure(self, monkeypatch): + rt = AgentRuntime() + + def boom(*a, **k): + raise RuntimeError("summarizer exploded") + + monkeypatch.setattr(rt, "run", boom) + agent = Agent( + name="support", + model="openai/gpt-4o", + semantic_memory=SemanticMemory(store=_FakeStore()), + ) + # Must not raise. + rt._maybe_save_conversation_memory( + agent, + AgentResult(status=Status.COMPLETED, messages=[{"role": "user", "content": "hi"}]), + None, + ) + + def test_apply_retrieval_prepends_context_without_mutating_original(self): + rt = AgentRuntime() + store = _FakeStore() + sm = SemanticMemory(store=store) + sm.get_context = lambda q: "Relevant context from memory:\n 1. plan=enterprise" # type: ignore + agent = Agent( + name="support", model="openai/gpt-4o", instructions="Be helpful.", semantic_memory=sm + ) + + augmented = rt._apply_memory_retrieval(agent, "what plan?") + assert augmented is not agent + assert augmented.instructions.startswith("Relevant context from memory:") + assert "Be helpful." in augmented.instructions + assert agent.instructions == "Be helpful." # original untouched + + def test_apply_retrieval_noop_without_memory(self): + rt = AgentRuntime() + agent = Agent(name="plain", model="openai/gpt-4o", instructions="Hi") + assert rt._apply_memory_retrieval(agent, "q") is agent + + +class TestSummaryHelpers: + def test_agent_model_str_fallback(self): + assert _agent_model_str(Agent(name="a", model="openai/gpt-4o")) == "openai/gpt-4o" + + def test_parse_summary_from_dict(self): + s, f, t = _parse_summary_output({"summary": "x", "facts": ["a"], "tags": ["b"]}) + assert s == "x" and f == ["a"] and t == ["b"] + + def test_parse_summary_from_wrapped_result(self): + s, f, t = _parse_summary_output({"result": {"summary": "y", "facts": [], "tags": []}}) + assert s == "y" + + def test_agent_stores_memory_attrs(self): + sm = SemanticMemory(store=_FakeStore()) + agent = Agent(name="a", model="openai/gpt-4o", semantic_memory=sm) + assert agent.semantic_memory is sm + assert agent.memory_summary_model is None # defaults to None -> reuse agent model + assert agent.feedback_sink is None diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/AgentCompiler.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/AgentCompiler.java index 11ed7682..80901c64 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/AgentCompiler.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/AgentCompiler.java @@ -40,6 +40,18 @@ public class AgentCompiler { "message", "${workflow.input.prompt}", "media", "${workflow.input.media}"); + /** + * Distillation prompt for the long-term (OCG) memory summarizer LLM step. + * Kept in sync with the Python {@code MEMORY_SUMMARIZER_INSTRUCTIONS} in + * {@code ocg_memory.py}. + */ + private static final String MEMORY_SUMMARIZER_INSTRUCTIONS = + "You distill a conversation into a durable memory. Read the transcript and " + + "extract only reusable, durable facts about the user, their preferences, and " + + "the task — the kind of thing worth remembering for next time. Ignore greetings, " + + "filler, and one-off details. Write a one-paragraph summary, a short list of " + + "facts, and a few topical tags. Be concise and concrete."; + private int timeoutSeconds = 0; private int llmRetryCount = 3; private int contextMaxSizeBytes = 32768; @@ -661,6 +673,11 @@ WorkflowDef compileWithTools(AgentConfig config) { // have null content on the first loop iteration. initVars.put("_human_feedback", ""); } + if (config.getLongTermMemory() != null) { + // Pre-initialize to empty string so the LTM system message has + // non-null content before the ltm_search/format tasks run. + initVars.put("_ltm_context", ""); + } WorkflowTask initState = new WorkflowTask(); initState.setType("SET_VARIABLE"); initState.setTaskReferenceName(toRef(config.getName()) + "_init_state"); @@ -670,6 +687,16 @@ WorkflowDef compileWithTools(AgentConfig config) { // Prefill tool calls: execute before the loop so results are in LLM context allTasks.addAll(prefill.tasks()); + // ── Long-term (OCG) memory: retrieve + inject (pre-loop) ───────── + // Search OCG for relevant memories, format them into a text block, + // and stash it in the ``_ltm_context`` workflow variable so buildLlmTask's + // system message picks it up on every turn. Best-effort: the search/format + // tasks are optional so a memory outage never fails the agent. No-op when + // longTermMemory is absent. + if (config.getLongTermMemory() != null) { + allTasks.addAll(buildLtmRetrievalTasks(config)); + } + // Required tools enforcement: wrap loop + check in outer DO_WHILE if (config.getRequiredTools() != null && !config.getRequiredTools().isEmpty()) { String checkRef = toRef(config.getName()) + "_required_tools_check"; @@ -702,10 +729,15 @@ WorkflowDef compileWithTools(AgentConfig config) { } // Post-loop: resolve output (guardrail fix or human edit may override LLM output) + // ``finalOutputRef`` captures the JSONPath to the agent's final text result so + // the long-term memory distill step (below) can summarize it. Differs between + // the guardrail branch (resolve_output) and the non-guardrail branch (synth_output). + String finalOutputRef; List outGuardrails = getOutputGuardrails(config); if (!outGuardrails.isEmpty()) { String resolveRef = toRef(config.getName()) + "_resolve_output"; allTasks.add(buildResolveOutputTask(resolveRef, llmRef)); + finalOutputRef = resolveRef + ".output.result.result"; Map outputParams = new LinkedHashMap<>(); outputParams.put("result", ref(resolveRef + ".output.result.result")); @@ -724,6 +756,7 @@ WorkflowDef compileWithTools(AgentConfig config) { // their content arg). String synthRef = toRef(config.getName()) + "_synth_output"; allTasks.add(buildSynthesizeOutputTask(synthRef, llmRef)); + finalOutputRef = synthRef + ".output.result"; Map outputParams = new LinkedHashMap<>(); outputParams.put("result", ref(synthRef + ".output.result")); @@ -733,11 +766,267 @@ WorkflowDef compileWithTools(AgentConfig config) { wf.setOutputParameters(outputParams); } + // ── Long-term (OCG) memory: distill + save + feedback (post-loop) ── + // Runs AFTER the output-synthesis task so the distiller can summarize the + // agent's final result. All tasks are best-effort (optional=true) so a + // memory/feedback failure never fails the agent workflow. No-op when + // longTermMemory is absent. + if (config.getLongTermMemory() != null) { + allTasks.addAll(buildLtmSaveTasks(config, finalOutputRef)); + } + wf.setTasks(allTasks); applyTimeout(wf, config); return wf; } + // ── Long-term (OCG) memory compilation ────────────────────────────── + // Compiled only into compileWithTools() (the CE orchestrator path). + // compileSimple/compileHybrid are NOT yet covered. + + /** + * Build the pre-loop retrieval tasks for long-term (OCG) memory: + *
    + *
  1. {@code *_ltm_search} — HTTP POST {@code /api/v1/memories/search} + * (feedback-blended ranking).
    2. + *
  2. {@code *_ltm_format} — INLINE (GraalJS) that formats the hits into a + * text block (folding the good/bad signal) per + * {@link JavaScriptBuilder#formatMemorySearchScript()}.
    3. + *
  3. {@code *_ltm_set_context} — SET_VARIABLE stashing the formatted block + * into {@code _ltm_context} for the LLM system message.
    4. + *
+ * All tasks are {@code optional=true} (best-effort). + */ + List buildLtmRetrievalTasks(AgentConfig config) { + LongTermMemoryConfig ltm = config.getLongTermMemory(); + String base = toRef(config.getName()); + List tasks = new ArrayList<>(); + + // 1. Search HTTP task + String searchRef = base + "_ltm_search"; + Map searchBody = new LinkedHashMap<>(); + searchBody.put("query", "${workflow.input.prompt}"); + searchBody.put("agent", ltm.getAgent()); + searchBody.put("limit", ltm.getMaxResults() != null ? ltm.getMaxResults() : 5); + searchBody.put("include_shared", true); + if (ltm.getUser() != null && !ltm.getUser().isBlank()) { + searchBody.put("user", ltm.getUser()); + } + WorkflowTask searchTask = buildMemoryHttpTask( + searchRef, ltm.getOcgUrl() + "/api/v1/memories/search", "POST", ltm.getCredential(), searchBody); + tasks.add(searchTask); + + // 2. Format INLINE task + String formatRef = base + "_ltm_format"; + WorkflowTask formatTask = new WorkflowTask(); + formatTask.setType("INLINE"); + formatTask.setTaskReferenceName(formatRef); + formatTask.setOptional(true); + Map formatInputs = new LinkedHashMap<>(); + formatInputs.put("evaluatorType", "graaljs"); + formatInputs.put("expression", JavaScriptBuilder.formatMemorySearchScript()); + formatInputs.put("memories", "${" + searchRef + ".output.response.body.memories}"); + formatTask.setInputParameters(formatInputs); + tasks.add(formatTask); + + // 3. SET_VARIABLE: stash formatted block into _ltm_context + WorkflowTask setCtx = new WorkflowTask(); + setCtx.setType("SET_VARIABLE"); + setCtx.setTaskReferenceName(base + "_ltm_set_context"); + setCtx.setOptional(true); + setCtx.setInputParameters(Map.of("_ltm_context", "${" + formatRef + ".output.result}")); + tasks.add(setCtx); + + return tasks; + } + + /** + * Build the post-loop distill/save/feedback tasks for long-term (OCG) memory: + *
    + *
  1. {@code *_ltm_distill} — LLM_CHAT_COMPLETE that summarizes the ticket + + * final report into a {@link dev.agentspan.runtime.model.MemorySummary}-shaped + * JSON ({@code summary}, {@code facts}, {@code tags}).
    2. + *
  2. {@code *_ltm_build_value} — INLINE building the durable memory ``value`` + * string from summary + facts.
    3. + *
  3. {@code *_ltm_save} — HTTP POST {@code /api/v1/memories}.
    4. + *
  4. {@code *_ltm_feedback_links} — HTTP POST + * {@code /api/v1/memories/{key}/feedback-links}.
    5. + *
  5. {@code } — SIMPLE worker handing the links to the + * user's Python feedback_sink (only when {@code feedbackSink} is set).
    6. + *
+ * All tasks are {@code optional=true} (best-effort). + * + * @param finalOutputRef JSONPath (without ``${}``) to the agent's final result. + */ + List buildLtmSaveTasks(AgentConfig config, String finalOutputRef) { + LongTermMemoryConfig ltm = config.getLongTermMemory(); + String base = toRef(config.getName()); + List tasks = new ArrayList<>(); + + String scope = ltm.getScope() != null ? ltm.getScope() : "agent"; + // Stable per-conversation key. Mirrors the Python save which keys on + // ``conversation:{session_id or execution_id}``. + String memoryKey = "conversation:${workflow.workflowId}"; + + // 1. Distill LLM task — summarize the run into durable facts. + String distillRef = base + "_ltm_distill"; + WorkflowTask distillTask = buildMemoryDistillTask(distillRef, ltm.getSummaryModel(), finalOutputRef); + tasks.add(distillTask); + + // 2. Build value INLINE — summary + facts → durable value string. + String valueRef = base + "_ltm_build_value"; + WorkflowTask valueTask = new WorkflowTask(); + valueTask.setType("INLINE"); + valueTask.setTaskReferenceName(valueRef); + valueTask.setOptional(true); + Map valueInputs = new LinkedHashMap<>(); + valueInputs.put("evaluatorType", "graaljs"); + valueInputs.put("expression", JavaScriptBuilder.buildMemoryValueScript()); + // LLM_CHAT_COMPLETE exposes its result as a JSON *string* at output.result + // (jsonOutput only nudges the model; the server does not parse it). So pass + // the raw string and let the INLINE JSON.parse it — distillRef.output.result.summary + // would never resolve. summary/facts/tags are then read from THIS task below. + valueInputs.put("distilled", "${" + distillRef + ".output.result}"); + valueTask.setInputParameters(valueInputs); + tasks.add(valueTask); + + // 3. Save HTTP task. + String saveRef = base + "_ltm_save"; + Map saveBody = new LinkedHashMap<>(); + saveBody.put("key", memoryKey); + saveBody.put("agent", ltm.getAgent()); + saveBody.put("value", "${" + valueRef + ".output.result.value}"); + saveBody.put("description", "${" + valueRef + ".output.result.description}"); + saveBody.put("scope", scope); + saveBody.put("source", "agent_inferred"); + saveBody.put("tags", "${" + valueRef + ".output.result.tags}"); + if (ltm.getUser() != null && !ltm.getUser().isBlank()) { + saveBody.put("user", ltm.getUser()); + } + WorkflowTask saveTask = buildMemoryHttpTask( + saveRef, ltm.getOcgUrl() + "/api/v1/memories", "POST", ltm.getCredential(), saveBody); + tasks.add(saveTask); + + // 4. Feedback-links HTTP task (mint signed good/bad capability URLs). + String linksRef = base + "_ltm_feedback_links"; + StringBuilder linksUri = new StringBuilder(); + linksUri.append(ltm.getOcgUrl()) + .append("/api/v1/memories/") + .append(memoryKey) + .append("/feedback-links?agent=") + .append(ltm.getAgent()); + if (ltm.getUser() != null && !ltm.getUser().isBlank()) { + linksUri.append("&user=").append(ltm.getUser()); + } + WorkflowTask linksTask = + buildMemoryHttpTask(linksRef, linksUri.toString(), "POST", ltm.getCredential(), null); + tasks.add(linksTask); + + // 5. feedback_sink SIMPLE worker — hand the links to the user's Python sink. + if (config.getFeedbackSink() != null && config.getFeedbackSink().getTaskName() != null) { + WorkflowTask sinkTask = new WorkflowTask(); + sinkTask.setName(config.getFeedbackSink().getTaskName()); + sinkTask.setTaskReferenceName(base + "_feedback_sink"); + sinkTask.setType("SIMPLE"); + sinkTask.setOptional(true); + Map sinkInputs = new LinkedHashMap<>(); + sinkInputs.put("memory_key", memoryKey); + sinkInputs.put("summary", "${" + valueRef + ".output.result.summary}"); + sinkInputs.put("facts", "${" + valueRef + ".output.result.facts}"); + sinkInputs.put("tags", "${" + valueRef + ".output.result.tags}"); + sinkInputs.put("good_url", "${" + linksRef + ".output.response.body.good_url}"); + sinkInputs.put("bad_url", "${" + linksRef + ".output.response.body.bad_url}"); + sinkInputs.put("expires_at", "${" + linksRef + ".output.response.body.expires_at}"); + sinkInputs.put("agent", ltm.getAgent()); + if (ltm.getUser() != null && !ltm.getUser().isBlank()) { + sinkInputs.put("user", ltm.getUser()); + } + sinkTask.setInputParameters(sinkInputs); + tasks.add(sinkTask); + } + + return tasks; + } + + /** + * Build an HTTP task targeting the OCG BFF. The credential is written as a + * {@code ${NAME}} placeholder and rewritten by {@link ToolCompiler#escapeCredentialHeaders} + * into whatever the host resolves: embedded (orkes-conductor) → + * {@code ${workflow.secrets.NAME}} (host secret store — same path the OCG query + * tools use), standalone → {@code #{NAME}} (resolved by {@code CredentialAwareHttpTask} + * against the per-user execution token). Hardcoding {@code #{NAME}} here was wrong + * for the embedded host and produced 401s. Marked {@code optional=true} so memory + * failures never fail the agent. A {@code null} body sends no JSON body. + */ + WorkflowTask buildMemoryHttpTask( + String refName, String uri, String method, String credential, Map body) { + WorkflowTask task = new WorkflowTask(); + task.setName("http_ocg_memory"); + task.setTaskReferenceName(refName); + task.setType("HTTP"); + task.setOptional(true); + + Map httpReq = new LinkedHashMap<>(); + httpReq.put("uri", uri); + httpReq.put("method", method); + Map headers = new LinkedHashMap<>(); + headers.put("Authorization", "Bearer ${" + credential + "}"); + headers.put("Content-Type", "application/json"); + // Host-mode-aware rewrite: ${NAME} -> ${workflow.secrets.NAME} (embedded) or #{NAME} (standalone). + httpReq.put("headers", ToolCompiler.escapeCredentialHeaders(headers)); + httpReq.put("accept", "application/json"); + httpReq.put("connectionTimeOut", 30000); + httpReq.put("readTimeOut", 30000); + if (body != null) { + httpReq.put("body", body); + } + + Map inputs = new LinkedHashMap<>(); + inputs.put("http_request", httpReq); + // Forward execution token so CredentialAwareHttpTask resolves #{NAME} headers. + inputs.put("__agentspan_ctx__", "${workflow.input.__agentspan_ctx__}"); + task.setInputParameters(inputs); + return task; + } + + /** + * Build the LLM distillation task: summarize the ticket + final report into a + * MemorySummary-shaped JSON ({@code summary}, {@code facts}, {@code tags}). + * Forces JSON output. Marked {@code optional=true}. + */ + WorkflowTask buildMemoryDistillTask(String distillRef, String summaryModel, String finalOutputRef) { + ParsedModel parsed = ModelParser.parse(summaryModel); + WorkflowTask llm = new WorkflowTask(); + llm.setName("LLM_CHAT_COMPLETE"); + llm.setTaskReferenceName(distillRef); + llm.setType("LLM_CHAT_COMPLETE"); + llm.setOptional(true); + + String systemMessage = MEMORY_SUMMARIZER_INSTRUCTIONS + + "\n\nRespond with a JSON object matching this schema: " + + "{\"summary\": string (one short paragraph: what happened / what was learned), " + + "\"facts\": array of strings (durable, reusable facts about the user or task; no chit-chat), " + + "\"tags\": array of strings (short topical tags)}. Output only valid JSON, no other text."; + + List messages = new ArrayList<>(); + messages.add(Map.of("role", "system", "message", systemMessage)); + messages.add(Map.of( + "role", "user", + "message", "TICKET:\n${workflow.input.prompt}\n\nFINAL REPORT:\n${" + finalOutputRef + "}")); + + Map inputs = new LinkedHashMap<>(); + inputs.put("llmProvider", parsed.getProvider()); + inputs.put("model", parsed.getModel()); + inputs.put("messages", messages); + inputs.put("jsonOutput", true); + inputs.put("maxTokens", 2048); + inputs.put("temperature", 0); + // Forward execution token so per-user credential resolution works. + inputs.put("__agentspan_ctx__", "${workflow.input.__agentspan_ctx__}"); + llm.setInputParameters(inputs); + return llm; + } + // ── Hybrid: tools AND sub-agents ──────────────────────────────── WorkflowDef compileHybrid(AgentConfig config) { @@ -1136,6 +1425,13 @@ WorkflowDef createWorkflow(AgentConfig config) { wf.setTimeoutSeconds(60L); wf.setTimeoutPolicy(null); wf.setInputParameters(WORKFLOW_INPUTS); + // Default ``media`` to an empty list so ``${workflow.input.media}`` never + // resolves to null. The SDK/API start path (AgentService) already defaults + // this, but inbound-webhook starts bypass AgentService, leaving the user + // ChatMessage's media null — the upstream ChatCompleteTask then NPEs on + // ``getMedia().stream()`` while assembling multi-turn history. inputTemplate + // values are defaults only; a caller-supplied ``media`` still overrides. + wf.setInputTemplate(Map.of("media", List.of())); return wf; } @@ -1286,6 +1582,18 @@ WorkflowTask buildLlmTask( } } + // Long-term (OCG) memory: inject retrieved context as a system message. + // ``_ltm_context`` is computed pre-loop by the ``*_ltm_search`` HTTP task + // + ``*_ltm_format`` INLINE and stored as a workflow variable (empty string + // when nothing relevant is found, so this message is harmless). Mirrors the + // ``_human_feedback`` system-message pattern. No-op when longTermMemory is + // absent. + if (config.getLongTermMemory() != null) { + messages.add(Map.of( + "role", "system", + "message", "${workflow.variables._ltm_context}")); + } + // Memory messages if (config.getMemory() != null && config.getMemory().getMessages() != null) { messages.addAll(config.getMemory().getMessages()); diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/ToolCompiler.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/ToolCompiler.java index c2faf812..70f9fa56 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/ToolCompiler.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/compiler/ToolCompiler.java @@ -68,6 +68,16 @@ private static Map escapeCredentialPlaceholders(Map header return escaped; } + /** + * Host-mode-aware rewrite of {@code ${NAME}} credential placeholders in an HTTP + * headers map. Exposed for compiler-emitted HTTP tasks (e.g. long-term memory) + * that must resolve credentials the same way tool HTTP calls do — embedded hosts + * get {@code ${workflow.secrets.NAME}}, standalone gets {@code #{NAME}}. + */ + public static Map escapeCredentialHeaders(Map headers) { + return escapeCredentialPlaceholders(headers); + } + /** * Rewrite a {@code ${NAME}} credential placeholder to the inert transport form {@code #{NAME}}. * diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/controller/AgentController.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/controller/AgentController.java index 18e6c96a..17b55afe 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/controller/AgentController.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/controller/AgentController.java @@ -370,7 +370,8 @@ public List getTaskLogs(@PathVariable String taskId) { /** * Search executions (pass-through to Conductor search, used by UI). An optional * {@code classifier} filter (comma-separated) is folded into the query as - * {@code classifier IN (...)}. + * {@code classifier IN (...)}; {@code topLevelOnly=true} restricts results to root executions + * ({@code parentWorkflowId = ""}). */ @GetMapping("/executions/search") public SearchResult searchExecutionsRaw( @@ -379,8 +380,9 @@ public SearchResult searchExecutionsRaw( @RequestParam(defaultValue = "startTime:DESC") String sort, @RequestParam(required = false) String freeText, @RequestParam(required = false) String query, - @RequestParam(required = false) String classifier) { - return agentService.searchExecutionsRaw(start, size, sort, freeText, query, classifier); + @RequestParam(required = false) String classifier, + @RequestParam(required = false, defaultValue = "false") boolean topLevelOnly) { + return agentService.searchExecutionsRaw(start, size, sort, freeText, query, classifier, topLevelOnly); } // ── Bulk operations ───────────────────────────────────────────── diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/AgentConfig.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/AgentConfig.java index b49a3e85..ef4585ef 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/AgentConfig.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/AgentConfig.java @@ -56,6 +56,21 @@ public class AgentConfig { private List guardrails; private MemoryConfig memory; + /** + * Long-term (OCG-backed) memory configuration. When present, the compiler + * inlines memory retrieval (pre-loop) and distill/save/feedback (post-loop) + * steps into the workflow. Distinct from the short-term {@link #memory}. + */ + private LongTermMemoryConfig longTermMemory; + + /** + * Worker reference for the long-term memory {@code feedback_sink} callable. + * When present (and {@link #longTermMemory} is set), the compiler emits a + * post-loop SIMPLE task that hands the human good/bad capability links to + * the user's Python feedback_sink worker. + */ + private WorkerRef feedbackSink; + @Builder.Default private int maxTurns = 100; diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/LongTermMemoryConfig.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/LongTermMemoryConfig.java new file mode 100644 index 00000000..d806f489 --- /dev/null +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/model/LongTermMemoryConfig.java @@ -0,0 +1,60 @@ +/* + * Copyright (c) 2025 AgentSpan + * Licensed under the MIT License. See LICENSE file in the project root for details. + */ + +package dev.agentspan.runtime.model; + +import com.fasterxml.jackson.annotation.JsonInclude; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Long-term (OCG-backed) memory configuration DTO. + * + *

Emitted by the Python serializer when an {@code Agent} has a + * {@code semantic_memory} backed by an {@code OCGMemoryStore}. Drives the + * server-side compiler to inline memory retrieval (pre-loop) and + * distill/save/feedback (post-loop) steps into the Conductor workflow, so + * long-term memory works on the deployed/webhook execution path — not just + * the client-side {@code run()} wrapper. + * + *

Unlike short-term {@link MemoryConfig} (pre-loaded conversation + * messages), this drives HTTP calls to an OCG instance using a + * server-resolvable credential name (never the client token). + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@JsonInclude(JsonInclude.Include.NON_NULL) +public class LongTermMemoryConfig { + + /** Base URL of the OCG instance (no trailing slash). */ + private String ocgUrl; + + /** + * Server-resolvable credential NAME (e.g. {@code "OCG_PUBLIC_KEY"}) for the + * OCG bearer token. Resolved server-side via the {@code #{NAME}} placeholder + * in HTTP task headers — never the raw client token. + */ + private String credential; + + /** Agent owner / scope key, e.g. {@code "agent:ce-ticket-resolution"}. */ + private String agent; + + /** Optional user owner, e.g. {@code "user:alice"}. */ + private String user; + + /** Memory scope for writes (default {@code "agent"}). */ + private String scope; + + /** Max memories to retrieve per search. */ + private Integer maxResults; + + /** Model used by the distillation (memory summarizer) LLM step. */ + private String summaryModel; +} diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/service/AgentService.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/service/AgentService.java index 7e6e9d8c..1ae2222f 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/service/AgentService.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/service/AgentService.java @@ -1496,12 +1496,30 @@ public TaskListResponse getExecutionTasks(String executionId, String status, int public SearchResult searchExecutionsRaw( int start, int size, String sort, String freeText, String query) { - return searchExecutionsRaw(start, size, sort, freeText, query, null); + return searchExecutionsRaw(start, size, sort, freeText, query, null, false); } public SearchResult searchExecutionsRaw( int start, int size, String sort, String freeText, String query, String classifier) { - return workflowService.searchWorkflows(start, size, sort, freeText, withClassifierFilter(query, classifier)); + return searchExecutionsRaw(start, size, sort, freeText, query, classifier, false); + } + + /** + * Search executions with an optional {@code classifier} filter (folded in as + * {@code classifier IN (...)}) and an optional top-level-only restriction. Top-level executions + * are roots (no parent); roots store {@code parent_workflow_id = ""}, so the restriction is the + * filter {@code parentWorkflowId = ""}. Both are ANDed onto the caller's {@code query}. + */ + public SearchResult searchExecutionsRaw( + int start, int size, String sort, String freeText, String query, String classifier, boolean topLevelOnly) { + String effectiveQuery = withClassifierFilter(query, classifier); + if (topLevelOnly) { + String topLevelFilter = "parentWorkflowId = \"\""; + effectiveQuery = (effectiveQuery == null || effectiveQuery.isBlank()) + ? topLevelFilter + : effectiveQuery + " AND " + topLevelFilter; + } + return workflowService.searchWorkflows(start, size, sort, freeText, effectiveQuery); } /** diff --git a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/util/JavaScriptBuilder.java b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/util/JavaScriptBuilder.java index 9883d4ba..0c9908c7 100644 --- a/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/util/JavaScriptBuilder.java +++ b/server/conductor-agentspan/src/main/java/dev/agentspan/runtime/util/JavaScriptBuilder.java @@ -1863,4 +1863,77 @@ public static String extractJsonFenceScript() { // Nothing found + "return {plan_json: null, markdown_plan: text};"); } + + // ── Long-term (OCG) memory helpers ────────────────────────────────── + + /** + * Format OCG search hits into a system-message text block for injection + * into the agent's prompt. Reads {@code $.memories} (the + * {@code response.body.memories} array from the search HTTP task) and folds + * the human good/bad signal into each line — mirroring the Python + * {@code _with_signal}. Returns {@code ""} when there are no hits, so the + * injected system message is harmless when memory is empty. + */ + public static String formatMemorySearchScript() { + return iife(" var mems = $.memories;" + + " if (mems == null || !Array.isArray(mems) || mems.length === 0) { return ''; }" + + " var lines = ['Relevant context from memory:'];" + + " for (var i = 0; i < mems.length; i++) {" + + " var m = mems[i] || {};" + + " var content = m.value_preview || '';" + + " var good = parseInt(m.good_count || 0, 10) || 0;" + + " var bad = parseInt(m.bad_count || 0, 10) || 0;" + + " if (good || bad) {" + + " content += ' [good ' + good + ' / bad ' + bad + ']';" + + " var notes = m.feedback_notes || [];" + + " for (var j = 0; j < notes.length; j++) {" + + " var n = notes[j] || {};" + + " if (n.verdict === 'bad' && n.reason) {" + + " content += ' (bad: \"' + n.reason + '\")';" + + " }" + + " }" + + " }" + + " lines.push(' ' + (i + 1) + '. ' + content);" + + " }" + + " return lines.join('\\n');"); + } + + /** + * Parse the distiller LLM's JSON output and build the durable memory + * ``value`` string. The {@code LLM_CHAT_COMPLETE} task exposes its result as + * a JSON string at {@code output.result} (``jsonOutput`` only nudges + * the model to emit JSON; the server does not parse it), so this reads the + * raw string {@code $.distilled}, strips any prose/code-fence around the + * object, and {@code JSON.parse}s it. Mirrors the Python post-run save which + * appends a ``Facts:`` block. Returns {@code {value, description, summary, + * facts, tags}} so the save HTTP body and the feedback_sink task can read + * the parsed fields from this one task (the distiller's + * {@code output.result.summary} would never resolve — it is a string). + * Resilient: malformed JSON falls back to using the raw text as the summary. + */ + public static String buildMemoryValueScript() { + return iife(" var raw = $.distilled;" + + " var obj = {};" + + " if (raw != null && typeof raw === 'object') { obj = raw; }" + + " else if (typeof raw === 'string' && raw.length > 0) {" + + " var s = raw.trim();" + + " var f = s.indexOf('{'); var l = s.lastIndexOf('}');" + + " if (f >= 0 && l > f) { s = s.substring(f, l + 1); }" + + " try { obj = JSON.parse(s); } catch (e) { obj = {summary: raw}; }" + + " }" + + " var summary = obj.summary;" + + " if (summary == null) { summary = ''; }" + + " if (typeof summary !== 'string') { summary = String(summary); }" + + " var facts = Array.isArray(obj.facts) ? obj.facts : [];" + + " var tags = Array.isArray(obj.tags) ? obj.tags : [];" + + " var value = summary;" + + " if (facts.length > 0) {" + + " value += '\\n\\nFacts:\\n';" + + " var fl = [];" + + " for (var i = 0; i < facts.length; i++) { fl.push('- ' + facts[i]); }" + + " value += fl.join('\\n');" + + " }" + + " return {value: value, description: value.substring(0, 200)," + + " summary: summary, facts: facts, tags: tags};"); + } }