From Documents to Knowledge Graph: The Sifter Pipeline

async def _aextract_chunk(
    chunk: TextChunk,
    doc_id: str,
    llm: LLMClient,
    domain: DomainConfig,
    doc_context: str = "",
) -> ExtractionResult:
    """Extract entities and relations from a single chunk (async).

    Uses a combined prompt (1 LLM call) instead of separate entity + relation
    calls (2 LLM calls). Falls back to entity-only on parse failure.
    """
    prompt = build_combined_prompt(chunk.text, doc_id, domain, doc_context=doc_context)
    try:
        data = await llm.acall_json(prompt)
    except (RuntimeError, ValueError) as e:
        logger.warning(f"Extraction failed for {doc_id} chunk {chunk.chunk_index}: {e}")
        return ExtractionResult(source_document=doc_id, chunk_index=chunk.chunk_index)

    # Schema discovery for schema-free domains
    if domain.schema_free and output_dir is not None:
        from sift_kg.domains.discovery import (
            discover_domain,
            load_discovered_domain,
            save_discovered_domain,
        )

        discovered_path = output_dir / "discovered_domain.yaml"
        cached = load_discovered_domain(discovered_path)
        if cached is not None and not force:
            logger.info(f"Using cached discovered schema ({len(cached.entity_types)} entity types)")
            domain = cached
        else:
            samples = [chunks[0].text[:3000]]
            try:
                domain = await discover_domain(samples, llm, domain.system_context or "")
                save_discovered_domain(domain, discovered_path)
                logger.info(f"Discovered schema: {len(domain.entity_types)} entity types, {len(domain.relation_types)} relation types")
            except (RuntimeError, ValueError) as e:
                logger.warning(f"Schema discovery failed, falling back to schema-free extraction: {e}")

    # Entity name → ID lookup (for resolving relation endpoints)
    name_to_id: dict[str, str] = {}

    # Pre-dedup: merge near-identical entity names deterministically
    canonical_map = prededup_entities(extractions)

    # Pre-create canonical entities so relations always resolve
    for entity_type, (names, _fallback) in (domain_canonical_entities or {}).items():
        for name in names:
            eid = _make_entity_id(name, entity_type)
            kg.add_entity(
                entity_id=eid,
                entity_type=entity_type,
                name=name,
                confidence=1.0,
                source_documents=[],
                attributes={"canonical": True},
            )
            name_to_id[name.lower().strip()] = eid
            stats["canonical_created"] += 1

def find_merge_candidates(
    kg: KnowledgeGraph,
    llm: LLMClient,
    entity_types: list[str] | None = None,
    concurrency: int = 4,
    use_embeddings: bool = False,
    system_context: str = "",
) -> tuple[MergeFile, list[RelationReviewEntry]]:
    """Find entities that likely refer to the same real-world thing.

    Args:
        kg: Knowledge graph with entities
        llm: LLM client for similarity judgments
        entity_types: Types to resolve (default: all types except DOCUMENT)
        concurrency: Max concurrent LLM calls
        use_embeddings: Use semantic clustering instead of alphabetical batching
        system_context: Domain context to help LLM understand entity names

    Returns:
        Tuple of (MergeFile with DRAFT proposals, list of variant relation proposals)
    """
    return asyncio.run(
        _afind_merge_candidates(kg, llm, entity_types, concurrency, use_embeddings, system_context)
    )

    # Rewrite edges
    edges_to_rewrite = list(kg.graph.edges(data=True, keys=True))
    for source, target, key, data in edges_to_rewrite:
        new_source = valid_map.get(source, source)
        new_target = valid_map.get(target, target)

        if new_source != source or new_target != target:
            # Remove old edge
            kg.graph.remove_edge(source, target, key=key)

            # Skip self-loops
            if new_source == new_target:
                stats["self_loops_removed"] += 1
                continue

            # Add rewritten edge
            kg.graph.add_edge(new_source, new_target, key=key, **data)

    # Remove merged nodes
    for member_id in valid_map:
        if kg.graph.has_node(member_id):
            kg.graph.remove_node(member_id)
            stats["nodes_removed"] += 1

def run_pipeline(
    doc_dir: Path,
    model: str,
    domain: DomainConfig,
    output_dir: Path,
    max_cost: float | None = None,
    include_narrative: bool = True,
) -> Path:
    """Run the full pipeline: extract → build → narrate.

    Skips resolve/apply-merges (those require human review).

    Args:
        doc_dir: Directory containing documents
        model: LLM model string
        domain: Domain configuration
        output_dir: Output directory for all artifacts
        max_cost: Budget cap in USD
        include_narrative: Whether to generate narrative at the end

    Returns:
        Path to output directory
    """
    run_extract(doc_dir, model, domain, output_dir, max_cost=max_cost)
    run_build(output_dir, domain)

    if include_narrative:
        system_context = domain.system_context or ""
        run_narrate(output_dir, model, system_context=system_context, max_cost=max_cost)