# Multimodal Retrieval with Cryptographic Trust Between LlamaIndex Agents

> Build a two-agent LlamaIndex retrieval pipeline where an ingestion agent processes multimodal documents with the ImageBlock content-block API, a query agent retrieves answers, and an Ed25519 trust boundary enforces a signed handoff between them, all validated by a pytest harness that measures retrieval accuracy and proves the handoff is rejected when forged.

- Canonical URL: https://agentry.press/tutorial/multimodal-retrieval-pipeline-with-llamaindex-agentmesh-trust-layer/
- Type: Tutorial
- Published: 2026-05-18
- By: agentry
- Tags: llamaindex, multi-agent, multimodal, retrieval, trust

---

## Why this matters

A retrieval pipeline that ingests PDFs, screenshots, and diagrams alongside prose cannot rely on a single text-only embedding pass. LlamaIndex models mixed-modality input with content blocks: a `ChatMessage` carries a `blocks` list that can hold a `TextBlock` and an `ImageBlock` together, rather than the older workaround of stuffing base64 strings into plain text. The ingestion side of a multimodal pipeline uses this API to ask a vision LLM for a text summary of each image, which is then embedded for hybrid retrieval. Agentic systems that operate across perception modalities face out-of-distribution inputs by design, which is part of why a hybrid text-plus-image retrieval path is worth the extra ingestion step [2].

The second concern is trust between agents. When one agent hands work to another, nothing in a plain function call proves the caller is who it claims to be or is permitted to make the call. This tutorial builds a two-agent split — one agent owns the messy ingestion boundary, the other owns the clean query interface — and places a cryptographic trust boundary between them: the ingestion agent signs the handoff with an Ed25519 key, and the query side verifies that signature against a registry before it will touch the index.

The upstream `llama-index-agent-agentmesh` integration was introduced for LlamaIndex to provide exactly this as a managed layer [1]. At the time of writing its published releases (0.1.0 and 0.2.0) import `AgentRunner`, a class removed from `llama-index-core` in the 0.13 agent refactor, so the package cannot be installed against any current core version. Rather than depend on a broken package, this tutorial implements the same guarantee — identity, an allow-list, a credential lifetime, and signature verification — directly from the `cryptography` library's Ed25519 primitives. The pattern transfers to AgentMesh if and when its releases are fixed.

## Prerequisites

- Python 3.11 or 3.12
- Familiarity with LlamaIndex query engines and the `VectorStoreIndex` API
- Basic understanding of vector stores and embedding-based retrieval
- An OpenAI API key is needed only for the optional live LLM mode at the end. The pipeline and the entire test suite run offline with a local embedding model and no key.

## Setup

Install the required packages. The pipeline runs offline with a local embedding model, so the only mandatory extras beyond `llama-index-core` are the HuggingFace embedding integration and `cryptography` for the trust boundary. The OpenAI packages are needed only for the optional live mode.

```bash
uv pip install "llama-index-core==0.14.15" \
    llama-index-embeddings-huggingface \
    cryptography \
    pillow \
    pytest \
    llama-index-llms-openai \
    llama-index-embeddings-openai \
    llama-index-multi-modal-llms-openai
```

The offline pipeline needs no API key. Export one only if you intend to run the live LLM mode at the end.

```bash
# skip_execution_reason: illustrative; needs a real API key
export OPENAI_API_KEY="sk-your-key-here"
```

## Step 1: Synthetic multimodal corpus

Create a small corpus of text nodes and a synthetic image node. In a real pipeline these would come from a PDF parser or an image store. Here you generate them programmatically so the tutorial runs without external files.

```python
# filename: corpus.py
from llama_index.core.schema import TextNode, ImageNode
import base64, io
from PIL import Image, ImageDraw

def make_synthetic_image_b64(label: str) -> str:
    """Return a tiny PNG encoded as a base64 string."""
    img = Image.new("RGB", (64, 64), color=(30, 80, 160))
    draw = ImageDraw.Draw(img)
    draw.text((4, 24), label[:6], fill=(255, 255, 255))
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode()


TEXT_NODES = [
    TextNode(
        text="The Eiffel Tower is a wrought-iron lattice tower in Paris, France, "
             "completed in 1889 as the entrance arch for the 1889 World's Fair.",
        metadata={"source": "wiki_eiffel", "modality": "text"},
        id_="node_eiffel_text",
    ),
    TextNode(
        text="The Colosseum in Rome is an elliptical amphitheatre built between "
             "70 and 80 AD. It could hold between 50,000 and 80,000 spectators.",
        metadata={"source": "wiki_colosseum", "modality": "text"},
        id_="node_colosseum_text",
    ),
    TextNode(
        text="The Great Wall of China stretches over 21,196 kilometres and was "
             "built across multiple dynasties starting in the 7th century BC.",
        metadata={"source": "wiki_great_wall", "modality": "text"},
        id_="node_great_wall_text",
    ),
    TextNode(
        text="Machu Picchu is a 15th-century Inca citadel located in the Eastern "
             "Cordillera of southern Peru at an elevation of 2,430 metres.",
        metadata={"source": "wiki_machu_picchu", "modality": "text"},
        id_="node_machu_picchu_text",
    ),
]

IMAGE_NODES = [
    ImageNode(
        image=make_synthetic_image_b64("Eiffel"),
        metadata={"source": "img_eiffel", "modality": "image",
                  "caption": "Photograph of the Eiffel Tower at dusk"},
        id_="node_eiffel_image",
    ),
]

ALL_NODES = TEXT_NODES + IMAGE_NODES
```

Verify the corpus module loads and the nodes are constructed correctly.

```python
from corpus import ALL_NODES, TEXT_NODES, IMAGE_NODES

print(f"Text nodes : {len(TEXT_NODES)}")
print(f"Image nodes: {len(IMAGE_NODES)}")
print(f"Total nodes: {len(ALL_NODES)}")
print("corpus_ok")
```

## Step 2: Multimodal message for image ingestion

LlamaIndex represents mixed-modality input as a `ChatMessage` whose `blocks` list holds typed content blocks. To summarise an image you build a message with a `TextBlock` carrying the instruction and an `ImageBlock` carrying the image, then send it to a vision LLM. This replaces the older pattern of formatting base64 into a plain-text prompt template.

```python
# filename: mm_message.py
from llama_index.core.llms import ChatMessage, ImageBlock, TextBlock, MessageRole
from llama_index.core.prompts import PromptTemplate

# Instruction sent to the vision LLM alongside the image content block.
IMAGE_SUMMARY_INSTRUCTION = (
    "You are an expert archivist. "
    "Describe the following image in two to three sentences suitable for "
    "full-text search. Focus on identifiable landmarks, colours, and context."
)


def build_image_summary_message(image_b64: str) -> ChatMessage:
    """
    Build a multimodal ChatMessage carrying a text instruction and an image
    content block. This is the real content-block primitive for mixing
    modalities: a ChatMessage whose ``blocks`` list holds a TextBlock and an
    ImageBlock, rather than stuffing base64 into a plain string template.
    """
    return ChatMessage(
        role=MessageRole.USER,
        blocks=[
            TextBlock(text=IMAGE_SUMMARY_INSTRUCTION),
            ImageBlock(image=image_b64.encode("utf-8")),
        ],
    )


# Text-only query prompt used by the query agent.
QUERY_TMPL_STR = (
    "You are a helpful assistant with access to a knowledge base about "
    "world landmarks.\n"
    "Context:\n"
    "{context_str}\n\n"
    "Question: {query_str}\n"
    "Answer concisely in one to two sentences."
)

QUERY_TMPL = PromptTemplate(QUERY_TMPL_STR)
```

```python
from mm_message import build_image_summary_message, QUERY_TMPL

msg = build_image_summary_message("aGVsbG8=")  # base64 of "hello"
print("message blocks:", [type(b).__name__ for b in msg.blocks])
print("query tmpl vars:", QUERY_TMPL.template_vars)
print("templates_ok")
```

## Step 3: Ingestion agent

The ingestion agent builds a `VectorStoreIndex` from the corpus. For each image node it calls the vision LLM to generate a text summary (using the multimodal message from Step 2), then adds that summary as an additional `TextNode` so the vector index can embed it. In offline mode the LLM call is skipped and a stub summary is used. The embedding model defaults to a local model so the pipeline runs without an API key.

```python
# filename: ingestion_agent.py
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core.schema import TextNode
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.storage.index_store import SimpleIndexStore
from llama_index.core.vector_stores import SimpleVectorStore
from corpus import ALL_NODES, IMAGE_NODES
from mm_message import build_image_summary_message


def summarise_image_node_stub(image_node) -> str:
    """Stub summary used when no LLM key is available (offline / CI)."""
    caption = image_node.metadata.get("caption", "")
    return f"[Stub summary] {caption} Source: {image_node.metadata.get('source', '')}"


def summarise_image_node_live(image_node, llm) -> str:
    """Call the vision LLM to summarise an image node using a multimodal
    ChatMessage (text instruction + image content block)."""
    message = build_image_summary_message(image_node.image)
    response = llm.chat([message])
    return str(response.message.content)


def build_index(use_live_llm: bool = False, llm=None, embed_model=None):
    """
    Ingest ALL_NODES into a VectorStoreIndex.
    Image nodes get a companion TextNode with an LLM-generated (or stub) summary.
    Returns the index.
    """
    nodes_to_index = list(ALL_NODES)

    for img_node in IMAGE_NODES:
        if use_live_llm and llm is not None:
            summary_text = summarise_image_node_live(img_node, llm)
        else:
            summary_text = summarise_image_node_stub(img_node)

        summary_node = TextNode(
            text=summary_text,
            metadata={
                "source": img_node.metadata["source"] + "_summary",
                "modality": "image_summary",
                "parent_image_id": img_node.node_id,
            },
            id_=img_node.node_id + "_summary",
        )
        nodes_to_index.append(summary_node)

    storage_context = StorageContext.from_defaults(
        docstore=SimpleDocumentStore(),
        index_store=SimpleIndexStore(),
        vector_store=SimpleVectorStore(),
    )

    if embed_model is None:
        # Default to a local embedding model so the pipeline runs offline.
        from llama_index.embeddings.huggingface import HuggingFaceEmbedding
        embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

    index = VectorStoreIndex(
        nodes=nodes_to_index,
        storage_context=storage_context,
        embed_model=embed_model,
        show_progress=False,
    )
    return index
```

## Step 4: Query agent

The query agent wraps the index in a vector retriever and exposes `retrieve` and `answer`. Retrieval needs no LLM, so the response synthesizer is built lazily and only when `answer` is called — this keeps offline retrieval and the eval harness free of any LLM dependency.

```python
# filename: query_agent.py
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.response_synthesizers import get_response_synthesizer
from mm_message import QUERY_TMPL


class QueryAgent:
    """
    Thin wrapper around a vector retriever. Retrieval needs no LLM, so the
    response synthesizer is built lazily and only when ``answer`` is called.
    This keeps offline retrieval (and the eval harness) free of any LLM
    dependency.
    """

    def __init__(self, index, llm=None, similarity_top_k: int = 3):
        self._llm = llm
        self._retriever = VectorIndexRetriever(
            index=index,
            similarity_top_k=similarity_top_k,
        )

    def retrieve(self, question: str):
        """Return the raw retrieved nodes without synthesis (no LLM needed)."""
        return self._retriever.retrieve(question)

    def answer(self, question: str) -> str:
        """Return a synthesised answer string. Requires an LLM."""
        synth_kwargs = {"llm": self._llm} if self._llm is not None else {}
        response_synthesizer = get_response_synthesizer(
            response_mode="compact",
            text_qa_template=QUERY_TMPL,
            **synth_kwargs,
        )
        engine = RetrieverQueryEngine(
            retriever=self._retriever,
            response_synthesizer=response_synthesizer,
        )
        return str(engine.query(question))
```

## Step 5: Cryptographic trust boundary

The trust boundary gives each agent an Ed25519 identity and gates the handoff between them. The ingestion agent signs a handoff token; the receiving side verifies it against a registry of known public keys and an explicit allow-list of permitted caller-to-callee edges. A token from an unknown caller, over a disallowed edge, past its time-to-live, or carrying a tampered payload fingerprint is rejected with `TrustError`. Verification fails closed: anything not explicitly registered and allowed is denied.

```python
# filename: trust.py
"""
A minimal cryptographic trust boundary between agents, built from primitives.

Each agent owns an Ed25519 keypair (its identity). When one agent hands off
to another, it signs a handoff token. The receiving side verifies the token
against a registry of known public keys and an explicit allow-list of caller
to callee edges, and rejects tokens that are forged, unauthorised, or expired.
"""
from __future__ import annotations

import json
import os
import time
from dataclasses import dataclass

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

DEFAULT_TTL_SECONDS = 900  # 15-minute credential lifetime


class TrustError(Exception):
    """Raised when a handoff is unknown, unauthorised, expired, or forged."""


@dataclass(frozen=True)
class Handoff:
    """A signed inter-agent handoff token."""
    caller: str
    callee: str
    issued_at: float
    nonce: str
    fingerprint: str
    signature: bytes

    def signed_bytes(self) -> bytes:
        """Canonical bytes that the signature covers. Excludes the signature
        itself; field order is fixed so signer and verifier agree."""
        payload = {
            "caller": self.caller,
            "callee": self.callee,
            "issued_at": self.issued_at,
            "nonce": self.nonce,
            "fingerprint": self.fingerprint,
        }
        return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()


class AgentIdentity:
    """An agent's cryptographic identity: a name, declared capabilities, and
    an Ed25519 private key. The private key never leaves the agent; only the
    public key is registered with peers."""

    def __init__(self, name: str, capabilities: list[str], private_key: Ed25519PrivateKey):
        self.name = name
        self.capabilities = list(capabilities)
        self._private_key = private_key

    @classmethod
    def generate(cls, name: str, capabilities: list[str] | None = None) -> "AgentIdentity":
        return cls(name, capabilities or [], Ed25519PrivateKey.generate())

    @property
    def public_key(self) -> Ed25519PublicKey:
        return self._private_key.public_key()

    def sign_handoff(
        self, callee: str, fingerprint: str, issued_at: float | None = None
    ) -> Handoff:
        """Sign a handoff to ``callee``. ``fingerprint`` binds the token to the
        payload being handed off (e.g. a digest of the index) so a captured
        token can't be replayed against different data."""
        ts = time.time() if issued_at is None else issued_at
        nonce = os.urandom(16).hex()
        unsigned = Handoff(self.name, callee, ts, nonce, fingerprint, b"")
        signature = self._private_key.sign(unsigned.signed_bytes())
        return Handoff(self.name, callee, ts, nonce, fingerprint, signature)


class TrustRegistry:
    """Holds the public key of each known agent and the set of permitted
    caller to callee edges. Verification fails closed: anything not
    explicitly registered and allowed is rejected."""

    def __init__(self, ttl_seconds: int = DEFAULT_TTL_SECONDS):
        self._keys: dict[str, Ed25519PublicKey] = {}
        self._edges: set[tuple[str, str]] = set()
        self._ttl = ttl_seconds

    def register(self, identity: AgentIdentity) -> None:
        self._keys[identity.name] = identity.public_key

    def allow(self, caller: str, callee: str) -> None:
        self._edges.add((caller, callee))

    def verify(self, handoff: Handoff, expected_fingerprint: str) -> None:
        """Raise TrustError unless the handoff is from a known caller, over an
        allowed edge, within the TTL, carries the expected fingerprint, and
        bears a valid signature. Returns None on success."""
        pub = self._keys.get(handoff.caller)
        if pub is None:
            raise TrustError(f"unknown caller: {handoff.caller!r}")
        if (handoff.caller, handoff.callee) not in self._edges:
            raise TrustError(
                f"edge not allowed: {handoff.caller!r} -> {handoff.callee!r}"
            )
        if time.time() - handoff.issued_at > self._ttl:
            raise TrustError("handoff expired")
        if handoff.fingerprint != expected_fingerprint:
            raise TrustError("fingerprint mismatch (payload tampered)")
        try:
            pub.verify(handoff.signature, handoff.signed_bytes())
        except InvalidSignature as exc:
            raise TrustError("invalid signature") from exc
```

```python
from trust import AgentIdentity, TrustRegistry, TrustError

ingestion = AgentIdentity.generate("ingestion_agent", ["ingest"])
query = AgentIdentity.generate("query_agent", ["retrieve"])

registry = TrustRegistry()
registry.register(ingestion)
registry.register(query)
registry.allow(caller="ingestion_agent", callee="query_agent")

# A valid handoff is accepted.
handoff = ingestion.sign_handoff(callee="query_agent", fingerprint="abc123")
registry.verify(handoff, expected_fingerprint="abc123")
print("valid handoff accepted")

# The reverse direction is not an allowed edge.
try:
    registry.verify(query.sign_handoff(callee="ingestion_agent", fingerprint="x"),
                    expected_fingerprint="x")
except TrustError as exc:
    print("rejected reverse handoff:", exc)

print("trust_ok")
```

## Step 6: Pipeline orchestrator

The orchestrator ties ingestion and query together. After the ingestion agent builds the index it computes a fingerprint of the index contents, signs a handoff bound to that fingerprint, and the registry verifies it before the query agent is allowed to touch the index. The verification is enforced — a failed check raises `TrustError` and the pipeline stops, rather than logging a warning and continuing.

```python
# filename: pipeline.py
import hashlib

from ingestion_agent import build_index
from query_agent import QueryAgent
from trust import AgentIdentity, TrustRegistry, TrustError


def index_fingerprint(index) -> str:
    """A stable digest of the index contents, used to bind a handoff token to
    the exact data being handed off."""
    node_ids = sorted(index.index_struct.nodes_dict.keys())
    digest = hashlib.sha256("|".join(node_ids).encode()).hexdigest()
    return digest


def build_trust(ingestion: AgentIdentity, query: AgentIdentity) -> TrustRegistry:
    """Register both identities and permit only ingestion -> query."""
    registry = TrustRegistry()
    registry.register(ingestion)
    registry.register(query)
    registry.allow(caller="ingestion_agent", callee="query_agent")
    return registry


def run_pipeline(
    question: str,
    use_live_llm: bool = False,
    llm=None,
    embed_model=None,
    ingestion: AgentIdentity | None = None,
    query: AgentIdentity | None = None,
    registry: TrustRegistry | None = None,
) -> dict:
    """
    1. Ingestion agent builds the index.
    2. Ingestion signs a handoff; the registry verifies it before the query
       agent is allowed to touch the index. A failed check raises TrustError.
    3. Query agent retrieves and answers.
    Returns a dict with 'answer' and 'retrieved_sources'.
    """
    ingestion = ingestion or AgentIdentity.generate("ingestion_agent", ["ingest"])
    query = query or AgentIdentity.generate("query_agent", ["retrieve"])
    registry = registry or build_trust(ingestion, query)

    # --- Ingestion agent phase ---
    index = build_index(use_live_llm=use_live_llm, llm=llm, embed_model=embed_model)
    fingerprint = index_fingerprint(index)

    # --- Trust verification (enforced, not advisory) ---
    handoff = ingestion.sign_handoff(callee="query_agent", fingerprint=fingerprint)
    registry.verify(handoff, expected_fingerprint=fingerprint)  # raises TrustError on failure

    # --- Query agent phase ---
    agent = QueryAgent(index=index, llm=llm)
    nodes = agent.retrieve(question)
    answer = agent.answer(question) if use_live_llm else "(offline: synthesis skipped)"

    return {
        "answer": answer,
        "retrieved_sources": [n.node.metadata.get("source") for n in nodes],
        "retrieved_texts": [n.node.get_content()[:120] for n in nodes],
        "fingerprint": fingerprint,
    }
```

## Step 7: Pytest evaluation harness

The harness runs the pipeline offline (no key needed) and checks two things: that the correct source nodes are retrieved for each question, and that the trust boundary actually enforces the handoff — accepting a valid token and rejecting forged, unauthorized, and expired ones.

```python
# filename: test_pipeline.py
import time

import pytest

from pipeline import run_pipeline, build_trust, index_fingerprint, build_index
from trust import AgentIdentity, TrustRegistry, TrustError

# Each entry: (question, gold_source_substring)
EVAL_CASES = [
    ("When was the Eiffel Tower completed?", "wiki_eiffel"),
    ("How many spectators could the Colosseum hold?", "wiki_colosseum"),
    ("How long is the Great Wall of China?", "wiki_great_wall"),
    ("Where is Machu Picchu located?", "wiki_machu_picchu"),
    ("What is the Eiffel Tower made of?", "wiki_eiffel"),
]


@pytest.fixture(scope="module")
def pipeline_results():
    """Run all eval cases once and cache results."""
    results = []
    for question, gold_source in EVAL_CASES:
        result = run_pipeline(question, use_live_llm=False)
        results.append({
            "question": question,
            "gold_source": gold_source,
            "retrieved_sources": result["retrieved_sources"],
        })
    return results


@pytest.mark.parametrize("case_idx,question,gold_source", [
    (i, q, g) for i, (q, g) in enumerate(EVAL_CASES)
])
def test_retrieval_hit(pipeline_results, case_idx, question, gold_source):
    """Gold source must appear in the retrieved sources for this question."""
    sources = pipeline_results[case_idx]["retrieved_sources"]
    hit = any(gold_source in (s or "") for s in sources)
    assert hit, (
        f"Question: {question!r}\nExpected source containing {gold_source!r}\n"
        f"Got: {sources}"
    )


def test_retrieval_accuracy(pipeline_results):
    """Overall hit-rate must be at least 80%."""
    hits = sum(
        1 for r in pipeline_results
        if any(r["gold_source"] in (s or "") for s in r["retrieved_sources"])
    )
    accuracy = hits / len(pipeline_results)
    print(f"\nRetrieval accuracy: {accuracy:.0%} ({hits}/{len(pipeline_results)})")
    assert accuracy >= 0.80, f"Accuracy {accuracy:.0%} is below the 80% threshold"


# --- Trust boundary tests: the handoff must actually be enforced. ---

def _setup():
    ingestion = AgentIdentity.generate("ingestion_agent", ["ingest"])
    query = AgentIdentity.generate("query_agent", ["retrieve"])
    registry = build_trust(ingestion, query)
    return ingestion, query, registry


def test_valid_handoff_accepted():
    ingestion, _query, registry = _setup()
    fp = "deadbeef"
    handoff = ingestion.sign_handoff(callee="query_agent", fingerprint=fp)
    registry.verify(handoff, expected_fingerprint=fp)  # must not raise


def test_reverse_edge_rejected():
    """query_agent -> ingestion_agent is not an allowed edge."""
    _ing, query, registry = _setup()
    handoff = query.sign_handoff(callee="ingestion_agent", fingerprint="x")
    with pytest.raises(TrustError, match="edge not allowed"):
        registry.verify(handoff, expected_fingerprint="x")


def test_unknown_caller_rejected():
    """An identity that was never registered cannot hand off."""
    _ing, _q, registry = _setup()
    impostor = AgentIdentity.generate("ingestion_agent", ["ingest"])  # same name, different key
    handoff = impostor.sign_handoff(callee="query_agent", fingerprint="x")
    with pytest.raises(TrustError, match="invalid signature"):
        registry.verify(handoff, expected_fingerprint="x")


def test_tampered_fingerprint_rejected():
    ingestion, _q, registry = _setup()
    handoff = ingestion.sign_handoff(callee="query_agent", fingerprint="real")
    with pytest.raises(TrustError, match="fingerprint mismatch"):
        registry.verify(handoff, expected_fingerprint="tampered")


def test_expired_handoff_rejected():
    ingestion, _q, registry = _setup()
    old = time.time() - 10_000  # well past the 900s TTL
    handoff = ingestion.sign_handoff(callee="query_agent", fingerprint="x", issued_at=old)
    with pytest.raises(TrustError, match="expired"):
        registry.verify(handoff, expected_fingerprint="x")
```

## Verify it works

Run the full pytest suite. The local embedding model (`BAAI/bge-small-en-v1.5`) is downloaded on first run (about 130 MB), so this block may take up to 60 seconds.

```bash
cd /workspace && python -m pytest test_pipeline.py -v --tb=short 2>&1 | tail -30
```

For a quick smoke test without pytest, run the pipeline directly and print the retrieved sources and the handoff fingerprint.

```python
from pipeline import run_pipeline

result = run_pipeline("When was the Eiffel Tower completed?")
print("Sources    :", result["retrieved_sources"])
print("Fingerprint:", result["fingerprint"][:16])
print("smoke_test_ok")
```

## Live LLM mode (requires API key)

Once you have `OPENAI_API_KEY` set, swap in real models. The ingestion agent then sends the multimodal `ChatMessage` to a vision model for a real image summary, and the query agent synthesises answers. This block is skipped in the sandbox because it makes live API calls.

```python
# skip_execution_reason: makes live OpenAI API calls
import os
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from pipeline import run_pipeline

llm = OpenAI(model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])
embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    api_key=os.environ["OPENAI_API_KEY"],
)

result = run_pipeline(
    "What year was the Eiffel Tower built and what event prompted its construction?",
    use_live_llm=True,
    llm=llm,
    embed_model=embed_model,
)
print("Answer  :", result["answer"])
print("Sources :", result["retrieved_sources"])
```

## Troubleshooting

**`ModuleNotFoundError: No module named 'llama_index.embeddings.huggingface'`** The local embedding model needs the HuggingFace integration. Run `uv pip install llama-index-embeddings-huggingface`.

**`ImportError: cannot import name 'AgentRunner'` when installing `llama-index-agent-agentmesh`** The published AgentMesh releases (0.1.0 / 0.2.0) import a class removed from `llama-index-core` 0.13, so they cannot be installed against current core. This tutorial deliberately does not depend on that package; the trust boundary is implemented from `cryptography` primitives in `trust.py` instead.

**Local embedding model download times out** The `BAAI/bge-small-en-v1.5` model is about 130 MB. If your network is slow, pre-download it with `python -c "from llama_index.embeddings.huggingface import HuggingFaceEmbedding; HuggingFaceEmbedding(model_name='BAAI/bge-small-en-v1.5')"` before running pytest.

**Retrieval accuracy below 80%** With the local embedding model and stub image summaries this is unlikely, but if it happens, increase `similarity_top_k` in `QueryAgent.__init__` from `3` to `5`. The eval harness checks for source presence anywhere in the top-k list, so a larger k raises recall.

**`TrustError: edge not allowed`** The registry permits only `ingestion_agent -> query_agent`. This fires if you verify a handoff in the reverse direction, or one whose caller/callee pair was never passed to `registry.allow(...)`. Register the edge you intend to permit, and sign the handoff with the matching `caller`.

**`TrustError: invalid signature`** The caller name resolved to a registered public key, but the signature did not verify against it — usually because the handoff was signed by a different key than the one registered under that name (an impostor), or the token was altered after signing.

**`ImportError: llama-index-llms-openai package not found`** This only affects `QueryAgent.answer` and the live mode. Offline retrieval never needs an LLM. Install the package if you want synthesis: `uv pip install llama-index-llms-openai`.

**`PIL.Image` import error** The `pillow` package must be installed. Run `uv pip install pillow` and retry.

## Next steps

- Replace `SimpleVectorStore` with a persistent store such as Chroma or Qdrant so the index survives process restarts without re-ingestion.
- Extend the multimodal ingestion agent to handle PDF pages by rendering each page to a PNG with `pdf2image` and passing it through `build_image_summary_message`, then storing both the rendered image node and its summary node.
- Persist each agent's Ed25519 key (for example with `private_bytes` / `public_bytes`) and distribute public keys out of band, so identities survive restarts and peers can be registered ahead of time rather than generated per run.
- Add OpenTelemetry tracing around each agent phase to capture per-agent span latency and export it to a local OpenTelemetry Collector for debugging slow retrieval paths.
- Expand the eval harness with answer-quality checks using an LLM-as-judge pattern: after each `agent.answer()` call, score the response against a reference answer and assert a minimum semantic-similarity threshold.

## FAQ

### How does the Ed25519 trust boundary enforce trust between the two agents?

Each agent owns an Ed25519 keypair. When the ingestion agent hands the index to the query agent it signs a handoff token; the receiving side verifies the signature against a registry of known public keys and an explicit allow-list of caller-to-callee edges, and raises `TrustError` if the token is from an unknown caller, over a disallowed edge, expired, or tampered with.

### How does the multimodal ingestion agent handle image nodes?

The ingestion agent builds a multimodal `ChatMessage` whose `blocks` list holds a `TextBlock` instruction and an `ImageBlock` carrying the image, and sends it to a vision LLM to produce a text summary. That summary is stored as a companion `TextNode` alongside the image so the vector index can embed and retrieve it.

### What embedding model does the pipeline use for retrieval?

The pipeline defaults to a local embedding model (`BAAI/bge-small-en-v1.5`) via `llama-index-embeddings-huggingface` so it runs offline with no API key. For live mode it substitutes OpenAI's `text-embedding-3-small`.

### How is the pipeline validated in the pytest harness?

The harness runs five evaluation cases with known gold source documents and checks whether the correct source appears in the top-k retrieved results, with a minimum threshold of 80%. It also runs five trust tests asserting that a valid handoff is accepted and that forged, unauthorized, and expired handoffs are rejected.

### What is AgentMesh and why does this tutorial not use it directly?

AgentMesh is an upstream LlamaIndex integration that aims to provide cryptographic agent identity and trust-gated query engines. As of this writing its published releases import a class removed from `llama-index-core` 0.13, so it cannot be installed against current core; this tutorial implements the same guarantee from Ed25519 primitives instead.

## References

1. https://github.com/run-llama/llama_index/releases/tag/v0.14.15
2. https://arxiv.org/abs/2605.06522v1