feat(AIC-3106): add multimodal history support to graph().invoke()

README.md

@@ -594,6 +594,7 @@ uv run python main.py [example] [flag-key] [user-input]
 | --- | --- | --- |
 | `agent` *(default)* | `uv run python main.py agent` | `config()` via the global registry — switches providers without code changes |
 | `graph` | `uv run python main.py graph` | `graph()` multi-agent workflow driven by a LaunchDarkly agent graph flag |
+| `graph-history` | `uv run python main.py graph-history` | `graph().invoke()` with multimodal `history` forwarded to the root node |
 | `openai-only` | `uv run python main.py openai-only` | `config()` with a custom `Registry` restricted to OpenAI handlers |
 | `streaming` | `uv run python main.py streaming` | `config().stream()` — token-by-token output |
 

examples/graph_history.py

@@ -0,0 +1,100 @@
+"""
+Example: graph().invoke() with multimodal conversation history.
+
+Passes a `history` list containing an image content block to a graph flag. Only
+the root node receives the history; downstream nodes see it through the normal
+node-to-node data passing. The image is a generated solid red square, so the
+model naming the colour is the signal that the image actually reached the
+provider.
+
+Usage (via main.py):
+    python main.py graph-history <graph-flag-key> "<user input>"
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+from typing import Any
+
+import examples.register  # noqa: F401 – side-effect: populate global_registry
+from examples.utils import new_context, solid_color_png_base64, write_output
+from launchdarkly_ai_server import global_registry, graph
+
+IMAGE_BLOCK = {
+    "type": "image",
+    "source": {
+        "type": "base64",
+        "media_type": "image/png",
+        "data": solid_color_png_base64((255, 0, 0)),
+    },
+}
+
+COLOR_QUESTION = (
+    "What colour is the square in the image I shared? Answer with just the colour name."
+)
+
+# Two supported shapes: history that carries only context (the user turn arrives
+# as user_input), and history that already ends with the user turn (user_input
+# is empty).
+SCENARIOS: list[dict[str, Any]] = [
+    {
+        "name": "image-in-history + question as user_input",
+        "history": [{"role": "user", "content": [IMAGE_BLOCK]}],
+        "user_input": COLOR_QUESTION,
+    },
+    {
+        "name": "history ends with the user turn, empty user_input",
+        "history": [
+            {"role": "user", "content": "I am going to share an image with you."},
+            {"role": "assistant", "content": "Sure — go ahead and share it."},
+            {
+                "role": "user",
+                "content": [IMAGE_BLOCK, {"type": "text", "text": COLOR_QUESTION}],
+            },
+        ],
+        "user_input": "",
+    },
+]
+
+
+async def run(key: str, user_input: str) -> None:
+    failures: list[str] = []
+
+    for scenario in SCENARIOS:
+        response = await graph(key, registry=global_registry).invoke(
+            user_input or scenario["user_input"],
+            new_context(),
+            {"user_id": "user-123"},
+            history=scenario["history"],
+        )
+
+        text = str(
+            response.get("response", "")
+            if isinstance(response, dict)
+            else getattr(response, "response", "")
+        )
+        saw_color = bool(re.search(r"\bred\b", text, re.IGNORECASE))
+
+        tag = "SAW" if saw_color else "DID NOT see"
+        print(
+            f"[graph-history-check] {scenario['name']}: model {tag} the image from history",
+            file=sys.stderr,
+        )
+        if not saw_color:
+            failures.append(scenario["name"])
+            print(
+                f"[graph-history-check] response was: {text[:300]}",
+                file=sys.stderr,
+            )
+
+        print(json.dumps(response, indent=2, default=str))
+        write_output(response)
+
+    if failures:
+        raise RuntimeError(
+            "graph() did not forward history to the root node for: "
+            + ", ".join(failures)
+            + ". Before the history feature lands this is the expected result."
+        )

examples/utils.py

@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import base64
 import dataclasses
 import json
 import random
 import string
+import struct
+import zlib
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -56,3 +59,30 @@ def write_output(data: Any) -> None:
         json.dumps(data, indent=2, default=_default_encoder), encoding="utf-8"
     )
     print(f"Output written to output/{filename}")
+
+
+def _png_chunk(kind: bytes, data: bytes) -> bytes:
+    return (
+        struct.pack(">I", len(data))
+        + kind
+        + data
+        + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
+    )
+
+
+def solid_color_png_base64(rgb: tuple[int, int, int], size: int = 64) -> str:
+    """Encodes a solid-colour PNG as base64 for multimodal examples.
+
+    Generating the image avoids committing a binary fixture, and the colour is
+    the only thing the model can report back — which makes it a usable signal
+    for whether the image actually reached the provider.
+    """
+    raw = b"".join(b"\x00" + bytes(rgb) * size for _ in range(size))
+    ihdr = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)
+    png = (
+        b"\x89PNG\r\n\x1a\n"
+        + _png_chunk(b"IHDR", ihdr)
+        + _png_chunk(b"IDAT", zlib.compress(raw))
+        + _png_chunk(b"IEND", b"")
+    )
+    return base64.b64encode(png).decode("ascii")

main.py

@@ -9,6 +9,7 @@
     python main.py streaming                launch-darkly-documentation-summarizer "Summarise feature flags in 3 bullets"
     python main.py judge                    launch-darkly-documentation-summarizer "What is the LaunchDarkly AI SDK?"
     python main.py graph                    my-agent-graph "What is the LaunchDarkly AI SDK?"
+    python main.py graph-history            my-agent-graph ""
     python main.py openai-only              my-openai-flag "Tell me about feature flags"
     python main.py langchain                my-langchain-flag "Tell me about feature flags"
     python main.py claude-agents            launch-darkly-documentation-summarizer "What is the LaunchDarkly AI SDK?"
@@ -44,6 +45,7 @@
     "agent": "examples.agent",
     "streaming": "examples.streaming",
     "graph": "examples.graph_example",
+    "graph-history": "examples.graph_history",
     "conversation": "examples.conversation",
     "history": "examples.history",
     "judge": "examples.judge_example",

packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py

@@ -35,7 +35,9 @@
     ProviderHandler,
     SpanMessage,
     SpanMessagePart,
+    compose_history,
     config,
+    content_to_text,
     create_handler,
     end_span_once,
     end_unfinished_spans,
@@ -320,17 +322,6 @@ def cancel_open_spans() -> None:
 # ---------------------------------------------------------------------------
 
 
-def _format_history(history: list[dict[str, Any]] | None) -> str | None:
-    if not history:
-        return None
-    lines = []
-    for msg in history:
-        role = msg.get("role", "user")
-        content = msg.get("content", "")
-        lines.append(f"{role}: {content}")
-    return "Conversation History:\n\n" + "\n".join(lines)
-
-
 def build_prompt(
     config: AiConfigRep,
     user_input: str | None,
@@ -362,15 +353,100 @@ def build_prompt(
             f"{config_history}\n\n{safe_input}" if config_history else safe_input
         )
 
-    history_text = _format_history(history)
-    if history_text:
-        system_prompt = (
-            f"{system_prompt}\n\n{history_text}" if system_prompt else history_text
-        )
-
     return safe_input, system_prompt
 
 
+def _parse_message_content(content: Any, variables: dict[str, Any]) -> Any:
+    return parse_template(content, variables) if isinstance(content, str) else content
+
+
+def _config_conversation_turns(
+    config: AiConfigRep, variables: dict[str, Any]
+) -> list[dict[str, Any]]:
+    return [
+        {
+            "role": message.get("role"),
+            "content": _parse_message_content(message.get("content", ""), variables),
+        }
+        for message in (config.get("messages") or [])
+        if message.get("role") != "system"
+    ]
+
+
+def _to_anthropic_user_content(content: Any) -> Any:
+    if isinstance(content, str):
+        return content
+
+    blocks: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            blocks.append({"type": "text", "text": block.get("text", "")})
+        elif block.get("type") == "image":
+            source = block.get("source", {})
+            if source.get("type") == "url":
+                mapped_source = {"type": "url", "url": source.get("url", "")}
+            else:
+                mapped_source = {
+                    "type": "base64",
+                    "media_type": source.get("media_type", ""),
+                    "data": source.get("data", ""),
+                }
+            blocks.append({"type": "image", "source": mapped_source})
+    return blocks
+
+
+async def _to_streamed_prompt(
+    turns: list[dict[str, Any]],
+) -> AsyncGenerator[dict[str, Any], None]:
+    # The envelope ``type`` has to agree with the message role. The CLI reading this stream
+    # accepts an "assistant" envelope as a replayed turn, but every other envelope type is
+    # required to carry role "user" — an assistant turn sent as ``type: "user"`` is rejected
+    # outright with "Expected message role 'user', got 'assistant'".
+    for turn in turns:
+        role = turn.get("role")
+        content = turn.get("content", "")
+        if role == "assistant":
+            yield {
+                "type": "assistant",
+                "message": {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": content_to_text(content)}],
+                },
+                "parent_tool_use_id": None,
+            }
+        else:
+            yield {
+                "type": "user",
+                "message": {
+                    "role": "user",
+                    "content": _to_anthropic_user_content(content),
+                },
+                "parent_tool_use_id": None,
+            }
+
+
+def build_query_prompt(
+    config: AiConfigRep,
+    user_input: str | None,
+    variables: dict[str, Any],
+    history: list[dict[str, Any]] | None,
+    fallback_prompt: str,
+) -> str | AsyncGenerator[dict[str, Any], None]:
+    if not history:
+        return fallback_prompt
+
+    turns = compose_history(
+        history=history,
+        user_input=user_input,
+        config_messages=(
+            []
+            if config.get("instructions")
+            else _config_conversation_turns(config, variables)
+        ),
+    )
+    return _to_streamed_prompt(turns)
+
+
 def _opening_of(prompt: str, system_prompt: str | None) -> Opening:
     return Opening(
         system_instructions=system_prompt,
@@ -451,6 +527,7 @@ async def _call_impl(
         open_root_span: Any = span
 
         prompt, system_prompt = build_prompt(config, user_input, vs, history)
+        query_prompt = build_query_prompt(config, user_input, vs, history, prompt)
         if config.get("outputFormat"):
             schema_instr = f"Respond with valid JSON matching this schema:\n{json.dumps(config['outputFormat'])}"
             system_prompt = (
@@ -510,7 +587,7 @@ async def _call_impl(
             # Held in a variable so the finally below can aclose() it. A bare `return` inside
             # `async for` abandons the generator, and asyncio's finalizer then raises RuntimeError
             # when the generator is suspended inside a real await in the SDK.
-            gen = query(prompt=prompt, options=options)
+            gen = query(prompt=query_prompt, options=options)
             try:
                 async for message in gen:
                     record_conversation_id(span, message)
@@ -647,6 +724,7 @@ async def _stream_gen(
     parent = parent_context_of(span)
 
     prompt, system_prompt = build_prompt(config, user_input, variables, history)
+    query_prompt = build_query_prompt(config, user_input, variables, history, prompt)
     opening = _opening_of(prompt, system_prompt)
 
     native_tool_map, user_config_tools, native_tool_names = partition_tools(
@@ -699,7 +777,7 @@ async def _stream_gen(
         )
 
         full_output = ""
-        gen = query(prompt=prompt, options=options)
+        gen = query(prompt=query_prompt, options=options)
         async for message in gen:
             record_conversation_id(span, message)
             record_native_tools(span, message, capture_content, catalog)

packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py

@@ -32,6 +32,7 @@
 from launchdarkly_ai_claude_agents.handler import (
     _build_hooks,
     build_prompt,
+    build_query_prompt,
     build_tool_mcp,
     partition_tools,
 )
@@ -103,6 +104,7 @@ async def _run_query(
     graph_key: str,
     run_id: str,
     child_subagent_tools: list[Any],
+    history: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     import importlib
 
@@ -116,6 +118,9 @@ async def _run_query(
     wrapped = _wrap_native_tools(tool_handlers, ld_context, track_data)
 
     prompt, system_prompt = build_prompt(node.config, input_text, variables)
+    query_prompt = build_query_prompt(
+        node.config, input_text, variables, history, prompt
+    )
     native_tool_map, user_config_tools, native_tool_names = partition_tools(
         node.config.get("tools"), wrapped
     )
@@ -170,7 +175,7 @@ async def _run_query(
     # Bare `return` inside `async for` abandons the generator — Python's asyncio
     # finalizer later tries to aclose() it and may raise RuntimeError if the
     # generator is suspended inside a real await in the SDK (AIC-2950).
-    gen = query_fn(prompt=prompt, options=options)
+    gen = query_fn(prompt=query_prompt, options=options)
     try:
         async for message in gen:
             if isinstance(message, ResultMessage):
@@ -207,6 +212,7 @@ def to_claude_agents(
     async def invoke(
         input_text: str = "",
         variables: dict[str, Any] | None = None,
+        history: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         import importlib
 
@@ -335,6 +341,7 @@ async def _subagent_execute(
                     def_obj.key,
                     run_id,
                     root_child_tools,
+                    history,
                 )
             except Exception as exc:
                 if span: