Source: http://www.poma-ai.com/docs/pipelines/unstructured-to-opensearch

# The Missing Link Between Unstructured.io and Optimal Retrieval in OpenSearch

<ByAuthor />

**The short answer:** Unstructured.io gives you a typed, ordered element list; OpenSearch gives you a `knn_vector` field, native BM25, and a dedicated `hybrid` query type with its own normalization pipeline. Wired together with `chunk_by_title` and a flat index, they still lose the hierarchy Unstructured recovered, and OpenSearch's hybrid normalization has no way to tell a decontextualized fragment from a complete one. The missing link is POMA: `PrimeCut().ingest()` consumes the raw element list, emits hierarchy-preserving [chunksets](/learn/chunking/chunksets) into a portable `.poma` archive, and `poma.vektoria` indexes only `(vector, file_id, chunkset_index)` while content lives on a volume. Retrieval runs OpenSearch's own `knn` query, then `assemble()` reassembles prompt-ready cheatsheets.

## What each end of the pipeline actually provides

**Unstructured.io** (`partition_pdf` or the hosted API) returns a flat, ordered list of typed elements — `Title`, `NarrativeText`, `ListItem`, `Table`, `Image` — each with `element_id`, `text`, and `metadata` (`page_number`, `text_as_html` for tables, optionally `image_base64`). Element order is document order; nothing records heading depth. Details: [The Optimal Chunker for Unstructured.io](/optimal-chunker-unstructured).

**OpenSearch** forked from Elasticsearch 7.10 in 2021 and built its own k-NN plugin: `knn_vector` fields (HNSW via `nmslib` or `faiss`), native BM25 on analyzed text, and a dedicated `hybrid` query type fused by a search pipeline's normalization processor — a distinct mechanism from Elasticsearch's RRF `rank`. It ranks whatever you indexed. Details: [OpenSearch Chunking Strategy for RAG](/optimal-chunks-opensearch).

## The naive wiring, and where it breaks

The common recipe — run `chunk_by_title`, index each fragment's raw text and vector without a lineage field, query with the `hybrid` query type — breaks in a pair-specific way:

- **`chunk_by_title` fragments carry no lineage above their nearest `Title`.** A `Table` element split across a character-limit boundary becomes two OpenSearch documents that a `hybrid` query's normalization processor scores independently — it has no signal that the two hits are one table.
- **Normalization blends scores per query, not per source document.** If a naive pipeline groups fragments only by nearest `Title` and skips populating `file_id`, the search pipeline can normalize and combine scores across documents from *different* files in a multi-tenant index, corrupting the fused ranking.
- **Overlap, if still applied, duplicates spans into both the HNSW graph and the inverted index** — more memory, slower segment merges, `top_k` crowded with near-duplicates.

## The pipeline, end to end

```bash
pip install unstructured opensearch-py
```

```python
from unstructured.partition.pdf import partition_pdf
from unstructured.staging.base import elements_to_json
from poma import PrimeCut
from poma.vektoria import assemble, records_from_archive, Volume
from poma.embeddings import get_embedder
from opensearchpy import OpenSearch

# 1. Your existing Unstructured call — unchanged.
elements = partition_pdf(
    filename="contract.pdf",
    strategy="hi_res",
    infer_table_structure=True,            # populates metadata.text_as_html
    extract_image_block_types=["Image"],
    extract_image_block_to_payload=True,   # inline base64 → POMA can describe figures
)
elements_to_json(elements, filename="contract.unstructured.json")

# 2. The missing link — raw element list in, a portable .poma archive out.
client = PrimeCut()  # reads POMA_API_KEY
client.ingest("contract.unstructured.json", download_dir="archives", filename="contract.poma")

embedder = get_embedder("local:BAAI/bge-small-en-v1.5")
vol = Volume("s3://your-bucket/poma")
osc = OpenSearch(hosts=[{"host": "localhost", "port": 9200}])

osc.indices.create(index="poma", body={"mappings": {"properties": {
    "embedding": {"type": "knn_vector", "dimension": embedder.dims,
                  "method": {"name": "hnsw", "space_type": "cosinesimil", "engine": "faiss"}},
    "file_id": {"type": "keyword"},
    "chunkset_index": {"type": "integer"},
}}}, ignore=400)

# 3. Content-free ingest — the index holds only routing metadata; content lives on the volume.
records = records_from_archive("archives/contract.poma")  # -> list[Record] (id, text, payload)
for r in records:
    vol.write_doc(r.payload["file_id"], {"file_id": r.payload["file_id"],
                                          "chunks": r.payload["chunks"], "text": r.text})
    osc.index(index="poma", id=r.id, body={
        "embedding": embedder.embed([r.text])[0],
        "file_id": r.payload["file_id"],
        "chunkset_index": r.payload["chunkset_index"],
    })

# 4. Retrieval — OpenSearch's normal knn query, then assemble.
qv = embedder.embed(["early termination conditions"])[0]
res = osc.search(index="poma", body={"query": {"knn": {"embedding": {
    "vector": qv, "k": 10}}}})
context = assemble(res, volume=vol)  # -> [{"file_id", "content"}, ...]
```

Every chunkset carries `file_id` because it comes from `records_from_archive`, not from a fragment's nearest `Title` — so a `hybrid` query's normalization processor never has to guess which document a hit belongs to. `file_id`/`chunkset_index` return by default on every hit, no extra flag needed.

## Metadata mapping: POMA fields → OpenSearch primitives

| POMA field (Unstructured-sourced) | OpenSearch primitive | Notes |
| --- | --- | --- |
| chunkset id (`chunkset_uuid`) | document `_id` | stable across re-indexing |
| `to_embed` (tables spliced as `text_as_html`) | `knn_vector` field | embedded before index, never stored as analyzed text |
| `file_id` | `keyword` field | returned by default for `assemble()` |
| `chunkset_index` | `integer` field | returned by default for `assemble()` |
| `page` (from `metadata.page_number`), `depth`, `chunk_index`, full chunkset text | volume document | content-free — never indexed, fetched by `assemble()` |

## Frequently asked questions

### How do I get Unstructured.io elements into OpenSearch for RAG?

Save the element list with `elements_to_json`, ingest it with `PrimeCut`, and keep the `.poma` archive. Create a `knn_vector` mapping with `file_id`/`chunkset_index`, index documents from `records_from_archive`, write chunkset content to a volume, then retrieve with `osc.search(knn=...)` plus `assemble(res, volume=vol)` — no extra metadata flag needed.

### Why does chunk_by_title fragmentation break OpenSearch's hybrid search pipeline?

`chunk_by_title` fragments carry no lineage, so the `hybrid` query type's normalization processor scores a decontextualized fragment the same as a complete one, and can blend scores across unrelated files if `file_id` was never populated.

### What OpenSearch mapping fields should Unstructured chunks carry?

`embedding` as `knn_vector` (dimension matching your embedder, HNSW method), `file_id` as `keyword`, `chunkset_index` as `integer`. Page, depth, chunk index, and full text stay off the document and live on a volume.

### Do Unstructured's tables survive into OpenSearch's knn_vector index?

Yes — POMA splices `text_as_html` into the chunkset before embedding, so the vector represents the whole table. The table's HTML itself lives on the volume, fetched by `assemble()` at retrieval time.

### Does POMA's assemble() need an extra metadata flag for OpenSearch results?

No — OpenSearch returns the document body by default, unlike Pinecone or Milvus which need an explicit ask. `assemble()` auto-detects OpenSearch's result shape as-is.

## Related recipes

Same parser, different store: [Unstructured.io → Elasticsearch](/pipelines/unstructured-to-elasticsearch) · [Unstructured.io → Vespa](/pipelines/unstructured-to-vespa) · [Unstructured.io → Chroma](/pipelines/unstructured-to-chroma)

Foundations: [The Optimal Chunker for Unstructured.io](/optimal-chunker-unstructured) · [OpenSearch Chunking Strategy for RAG](/optimal-chunks-opensearch) · [All pipeline recipes](/pipelines/) · [RAG chunking guide](/guides/rag-chunking/)