popoto.extraction.claude

popoto.extraction.claude

Claude (Anthropic) memory-extraction provider.

Wraps the Anthropic Messages API behind the AbstractExtractionProvider interface. Requires the anthropic package (pip install popoto[anthropic]).

Opt-in only: import popoto and import popoto.extraction never import anthropic -- this module is imported lazily by callers who want LLM-based extraction, mirroring popoto.embeddings.openai's pattern.

Example

from popoto.extraction.claude import ClaudeExtractionProvider provider = ClaudeExtractionProvider(api_key="your-key") facts = provider.extract("Alice met Bob at the conference in Paris.")

EXTRACTION_MODEL = 'claude-opus-4-8' module-attribute

Pinned model for Claude-based extraction. Not user-configurable.

EXTRACTION_MAX_TOKENS = 4096 module-attribute

Pinned max_tokens for Claude-based extraction. Not user-configurable.

EXTRACTION_PROMPT = 'You are a memory-extraction engine for an AI agent. Given a block of text (typically an LLM\'s response during a conversation), extract the discrete, independently-useful facts it contains.\n\nFor each fact:\n- "text": a short, self-contained statement of the fact, written so it makes sense on its own without the surrounding conversation.\n- "entities": the proper nouns / named entities mentioned in the fact (people, places, organizations, products, dates treated as named references, etc.). Use an empty list if none are present.\n- "importance": a score from 0.0 to 1.0 for how important this fact is likely to be to remember later (0.0 = trivial/filler, 1.0 = critical, load-bearing information).\n- "confidence": a score from 0.0 to 1.0 for how certain the source text is asserting this fact (0.0 = highly speculative or hedged, 1.0 = stated as definite fact).\n\nOnly extract facts that carry real informational content -- skip greetings, filler, and purely conversational scaffolding. If the text contains no extractable facts, return an empty "facts" array.' module-attribute

Pinned system prompt for Claude-based extraction. Not user-configurable.

FACTS_SCHEMA = {'type': 'object', 'properties': {'facts': {'type': 'array', 'items': {'type': 'object', 'properties': {'text': {'type': 'string'}, 'entities': {'type': 'array', 'items': {'type': 'string'}}, 'importance': {'type': 'number'}, 'confidence': {'type': 'number'}}, 'required': ['text', 'entities', 'importance', 'confidence'], 'additionalProperties': False}}}, 'required': ['facts'], 'additionalProperties': False} module-attribute

JSON schema for structured extraction output. Not user-configurable.

ClaudeExtractionProvider

Bases: AbstractExtractionProvider

Claude (Anthropic) memory-extraction provider.

Opt-in provider that calls the Anthropic Messages API once per extract() call, using structured JSON-schema output (output_config.format) to get typed facts back directly -- no tool_use, no assistant-turn prefill (which 400s on this model family).

Model and prompt are pinned module constants (EXTRACTION_MODEL, EXTRACTION_PROMPT), not constructor kwargs, per this project's experimental-constant convention.

On any API or parse failure, extract() logs a warning and returns an empty list -- it never raises, so a flaky extraction call never crashes the caller's turn loop.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	Anthropic API key. If None, reads from the ANTHROPIC_API_KEY env var (via the anthropic SDK's normal resolution).	None

Raises:

Type	Description
ImportError	If the anthropic package is not installed.

Source code in src/popoto/extraction/claude.py

class ClaudeExtractionProvider(AbstractExtractionProvider):
    """Claude (Anthropic) memory-extraction provider.

    Opt-in provider that calls the Anthropic Messages API once per
    ``extract()`` call, using structured JSON-schema output
    (``output_config.format``) to get typed facts back directly -- no
    tool_use, no assistant-turn prefill (which 400s on this model
    family).

    Model and prompt are pinned module constants (``EXTRACTION_MODEL``,
    ``EXTRACTION_PROMPT``), not constructor kwargs, per this project's
    experimental-constant convention.

    On any API or parse failure, ``extract()`` logs a warning and returns
    an empty list -- it never raises, so a flaky extraction call never
    crashes the caller's turn loop.

    Args:
        api_key: Anthropic API key. If None, reads from the
            ``ANTHROPIC_API_KEY`` env var (via the anthropic SDK's normal
            resolution).

    Raises:
        ImportError: If the ``anthropic`` package is not installed.
    """

    def __init__(self, api_key: Optional[str] = None):
        if not _anthropic_available:
            raise ImportError(
                "anthropic is required to use ClaudeExtractionProvider. "
                "Install it with: pip install popoto[anthropic]"
            )
        self._client = anthropic_module.Anthropic(api_key=api_key)

    def extract(self, text: str) -> List[ExtractedFact]:
        """Extract facts from text via one Claude API call.

        Args:
            text: Input text to extract facts from.

        Returns:
            List of ExtractedFact. Empty list if the text is empty/blank,
            no facts were found, or the API call/parse failed (a warning
            is logged in the failure case).
        """
        if not text or not text.strip():
            return []

        try:
            response = self._client.messages.create(
                model=EXTRACTION_MODEL,
                max_tokens=EXTRACTION_MAX_TOKENS,
                system=EXTRACTION_PROMPT,
                messages=[{"role": "user", "content": text}],
                output_config={
                    "format": {"type": "json_schema", "schema": FACTS_SCHEMA}
                },
            )
            raw_text = next(
                (block.text for block in response.content if block.type == "text"),
                None,
            )
            if raw_text is None:
                logger.warning("ClaudeExtractionProvider: no text block in response")
                return []
            parsed = json.loads(raw_text)
        except Exception as e:
            logger.warning("ClaudeExtractionProvider: extraction failed: %s", e)
            return []

        raw_facts = parsed.get("facts") if isinstance(parsed, dict) else None
        if not isinstance(raw_facts, list):
            logger.warning("ClaudeExtractionProvider: response missing 'facts' list")
            return []

        facts = []
        for raw_fact in raw_facts:
            if not isinstance(raw_fact, dict):
                continue
            fact_text = raw_fact.get("text")
            if not isinstance(fact_text, str) or not fact_text.strip():
                continue
            entities = raw_fact.get("entities")
            if not isinstance(entities, list):
                entities = []
            entities = [e for e in entities if isinstance(e, str)]
            importance = _clamp01(
                raw_fact.get("importance"), Defaults.EXTRACTION_DEFAULT_IMPORTANCE
            )
            confidence = _clamp01(
                raw_fact.get("confidence"), Defaults.EXTRACTION_DEFAULT_CONFIDENCE
            )
            facts.append(
                ExtractedFact(
                    text=fact_text.strip(),
                    entities=entities,
                    importance=importance,
                    confidence=confidence,
                )
            )
        return facts

extract(text)

Extract facts from text via one Claude API call.

Parameters:

Name	Type	Description	Default
text	str	Input text to extract facts from.	required

Returns:

Type	Description
List[ExtractedFact]	List of ExtractedFact. Empty list if the text is empty/blank,
List[ExtractedFact]	no facts were found, or the API call/parse failed (a warning
List[ExtractedFact]	is logged in the failure case).

Source code in src/popoto/extraction/claude.py

def extract(self, text: str) -> List[ExtractedFact]:
    """Extract facts from text via one Claude API call.

    Args:
        text: Input text to extract facts from.

    Returns:
        List of ExtractedFact. Empty list if the text is empty/blank,
        no facts were found, or the API call/parse failed (a warning
        is logged in the failure case).
    """
    if not text or not text.strip():
        return []

    try:
        response = self._client.messages.create(
            model=EXTRACTION_MODEL,
            max_tokens=EXTRACTION_MAX_TOKENS,
            system=EXTRACTION_PROMPT,
            messages=[{"role": "user", "content": text}],
            output_config={
                "format": {"type": "json_schema", "schema": FACTS_SCHEMA}
            },
        )
        raw_text = next(
            (block.text for block in response.content if block.type == "text"),
            None,
        )
        if raw_text is None:
            logger.warning("ClaudeExtractionProvider: no text block in response")
            return []
        parsed = json.loads(raw_text)
    except Exception as e:
        logger.warning("ClaudeExtractionProvider: extraction failed: %s", e)
        return []

    raw_facts = parsed.get("facts") if isinstance(parsed, dict) else None
    if not isinstance(raw_facts, list):
        logger.warning("ClaudeExtractionProvider: response missing 'facts' list")
        return []

    facts = []
    for raw_fact in raw_facts:
        if not isinstance(raw_fact, dict):
            continue
        fact_text = raw_fact.get("text")
        if not isinstance(fact_text, str) or not fact_text.strip():
            continue
        entities = raw_fact.get("entities")
        if not isinstance(entities, list):
            entities = []
        entities = [e for e in entities if isinstance(e, str)]
        importance = _clamp01(
            raw_fact.get("importance"), Defaults.EXTRACTION_DEFAULT_IMPORTANCE
        )
        confidence = _clamp01(
            raw_fact.get("confidence"), Defaults.EXTRACTION_DEFAULT_CONFIDENCE
        )
        facts.append(
            ExtractedFact(
                text=fact_text.strip(),
                entities=entities,
                importance=importance,
                confidence=confidence,
            )
        )
    return facts