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

README.md

@@ -541,6 +541,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
@@ -35,3 +38,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?"
@@ -43,6 +44,7 @@
     "agent": "examples.agent",
     "streaming": "examples.streaming",
     "graph": "examples.graph_example",
+    "graph-history": "examples.graph_history",
     "history": "examples.history",
     "judge": "examples.judge_example",
     "claude-agents": "examples.claude_agents_example",