KG Agent API Reference¶

Complete API reference for KnowledgeGraphAgent, the core class for querying Knowledge Packs.

KnowledgeGraphAgent¶

from wikigr.agent.kg_agent import KnowledgeGraphAgent

Constructor¶

KnowledgeGraphAgent(
    db_path: str,
    anthropic_api_key: str | None = None,
    read_only: bool = True,
    use_enhancements: bool = True,
    few_shot_path: str | None = None,
    enable_reranker: bool = True,
    enable_multidoc: bool = True,
    enable_fewshot: bool = True,
    enable_cross_encoder: bool = False,
    synthesis_model: str | None = None,
    cypher_pack_path: str | None = None,
    enable_multi_query: bool = False,
)

Parameters¶

Parameter	Type	Default	Description
`db_path`	`str`	(required)	Path to the LadybugDB database directory (e.g., `data/packs/go-expert/pack.db`)
`anthropic_api_key`	`str \\| None`	`None`	Anthropic API key. If `None`, reads from `ANTHROPIC_API_KEY` environment variable
`read_only`	`bool`	`True`	Open database in read-only mode. Enables concurrent access during expansion
`use_enhancements`	`bool`	`True`	Master switch for Phase 1 enhancements (reranking, multi-doc, few-shot). When `False`, all `enable_*` flags are ignored
`few_shot_path`	`str \\| None`	`None`	Path to few-shot examples JSON. When `None`, auto-detected from pack directory
`enable_reranker`	`bool`	`True`	Enable GraphReranker (degree centrality via RRF). Only active when `use_enhancements=True`
`enable_multidoc`	`bool`	`True`	Enable MultiDocSynthesizer (multi-article retrieval). Only active when `use_enhancements=True`
`enable_fewshot`	`bool`	`True`	Enable FewShotManager (example injection). Only active when `use_enhancements=True`
`enable_cross_encoder`	`bool`	`False`	Enable CrossEncoderReranker (joint query-document scoring). Opt-in. Only active when `use_enhancements=True`
`synthesis_model`	`str \\| None`	`None`	Claude model for synthesis. Defaults to `claude-opus-4-6`
`cypher_pack_path`	`str \\| None`	`None`	Path to OpenCypher expert pack for RAG-augmented Cypher generation
`enable_multi_query`	`bool`	`False`	Generate alternative query phrasings via Claude Haiku. Opt-in. When True, questions are sent to the Anthropic API

Example¶

# Default configuration (balanced)
agent = KnowledgeGraphAgent(
    db_path="data/packs/go-expert/pack.db",
)

# Maximum quality
agent = KnowledgeGraphAgent(
    db_path="data/packs/go-expert/pack.db",
    use_enhancements=True,
    enable_cross_encoder=True,
    enable_multi_query=True,
)

# Baseline (no enhancements)
agent = KnowledgeGraphAgent(
    db_path="data/packs/go-expert/pack.db",
    use_enhancements=False,
)

query()¶

def query(
    self,
    question: str,
    max_results: int = 10,
    use_graph_rag: bool = False,
) -> dict

Query the knowledge graph and synthesize an answer.

Parameters¶

Parameter	Type	Default	Description
`question`	`str`	(required)	The natural language question
`max_results`	`int`	`10`	Maximum number of vector search results. Clamped to [1, 1000]
`use_graph_rag`	`bool`	`False`	If True, delegate to `graph_query()` for multi-hop retrieval that follows LINKS_TO edges before synthesizing

Returns¶

A dictionary with the following structure:

{
    "answer": str,          # Synthesized answer text
    "sources": list[str],   # Article titles used as sources
    "entities": list[dict], # Entities mentioned in the answer
    "facts": list[str],     # Key facts from retrieved content
    "cypher_query": str,    # The vector search query executed
    "query_type": str,      # "vector_search" | "training_only_response"
    "token_usage": {
        "input_tokens": int,
        "output_tokens": int,
        "api_calls": int,
    },
}

query_type Values¶

Value	Meaning
`"vector_search"`	Normal path — pack content retrieved and used for synthesis. Also used when the vector index returns no results (the answer is synthesised from the model's training knowledge only).
`"training_only_response"`	Confidence gate fired — pack content was retrieved but similarity was below `CONTEXT_CONFIDENCE_THRESHOLD`. Claude answered using training knowledge only, without injecting pack context.

Example¶

result = agent.query("What is goroutine scheduling?")

print(result["answer"])
# "Go uses an M:N scheduling model where..."

print(result["sources"])
# ["runtime_scheduling", "goroutines"]

print(result["query_type"])
# "vector_search"  or "training_only_response" if confidence gate fired

print(result["token_usage"])
# {"input_tokens": 2847, "output_tokens": 312, "api_calls": 2}

Class Constants¶

Constant	Type	Default	Description
`DEFAULT_MODEL`	`str`	`"claude-opus-4-6"`	Default synthesis model
`VECTOR_CONFIDENCE_THRESHOLD`	`float`	`0.6`	Pre-defined constant for retrieval-layer filtering (not used in `query()` path)
`CONTEXT_CONFIDENCE_THRESHOLD`	`float`	`0.5`	Minimum cosine similarity required before pack content is injected into synthesis
`PLAN_CACHE_MAX_SIZE`	`int`	`128`	Intended maximum entries in the query plan cache. Not currently enforced — `_plan_cache` is an unbounded `dict` at runtime; this constant is defined but not referenced in any size-limiting code path.
`MAX_ARTICLE_CHARS`	`int`	`3000`	Maximum characters per article in synthesis context
`PLAN_MAX_TOKENS`	`int`	`512`	Maximum tokens for query planning
`SYNTHESIS_MAX_TOKENS`	`int`	`1024`	Maximum tokens for answer synthesis
`SEED_EXTRACT_MAX_TOKENS`	`int`	`256`	Maximum tokens for seed extraction
`CONTENT_QUALITY_THRESHOLD`	`float`	`0.3`	Minimum quality score for section inclusion in synthesis context
`STOP_WORDS`	`frozenset[str]`	~80 words	Common English function words excluded from keyword overlap scoring

Enhancement Module Classes¶

GraphReranker¶

from wikigr.agent.reranker import GraphReranker

reranker = GraphReranker(conn: kuzu.Connection)

# Calculate degree centrality for specific articles
centrality = reranker.calculate_centrality(
    article_ids: list[int],
) -> dict[int, float]

# Rerank results using weighted combination of vector similarity and centrality
reranked = reranker.rerank(
    vector_results: list[dict],  # [{"article_id": int, "score": float, ...}, ...]
    vector_weight: float = 0.6,  # vector similarity weight
    graph_weight: float = 0.4,   # degree centrality weight
) -> list[dict]

MultiDocSynthesizer¶

from wikigr.agent.multi_doc_synthesis import MultiDocSynthesizer

synthesizer = MultiDocSynthesizer(conn: kuzu.Connection)

# Expand seed articles by traversing LINKS_TO edges (BFS)
expanded = synthesizer.expand_to_related_articles(
    seed_articles: list[int],
    max_hops: int = 1,
    max_articles: int = 50,
) -> dict[int, dict]

# Create markdown text with numbered citations
text = synthesizer.synthesize_with_citations(
    articles: dict[int, dict],
    query: str,
) -> str

FewShotManager¶

from wikigr.agent.few_shot import FewShotManager

manager = FewShotManager(
    examples_path: str,  # path to JSON or JSONL file
)

examples = manager.find_similar_examples(
    query: str,
    k: int = 3,
) -> list[dict]
# Returns list of dicts with "score" key, sorted by similarity (descending)

CrossEncoderReranker¶

from wikigr.agent.cross_encoder import CrossEncoderReranker

reranker = CrossEncoderReranker()
# Downloads ms-marco-MiniLM-L-12-v2 (~33MB) on first use

reranked = reranker.rerank(
    query: str,
    results: list[dict],  # must have "content" or "title" key
    top_k: int = 5,
) -> list[dict]
# Each result gains a "ce_score" key (float, higher = more relevant)

If the model fails to load, rerank() returns results unchanged (passthrough).

Token Usage Tracking¶

The agent tracks cumulative token usage across all API calls:

agent = KnowledgeGraphAgent(db_path="pack.db")

result1 = agent.query("First question")
result2 = agent.query("Second question")

# Cumulative usage
print(agent.token_usage)
# {"input_tokens": 5694, "output_tokens": 624, "api_calls": 4}

# Per-query usage
print(result2["token_usage"])
# {"input_tokens": 2847, "output_tokens": 312, "api_calls": 2}

Error Handling¶

query() raises exceptions for infrastructure failures and lets programming bugs propagate. Only intentional degradation paths produce a result dict rather than an exception.

Exceptions raised by `query()`¶

Exception	Trigger	Retry?
`anthropic.APIConnectionError`	Network failure reaching Anthropic API	Yes — transient
`anthropic.APITimeoutError`	Anthropic API call timed out	Yes — transient
`anthropic.APIStatusError`	HTTP 4xx/5xx from Anthropic API	Depends on `status_code`
`RuntimeError`	LadybugDB database error	No — check pack integrity
`RuntimeError` or `OSError`	Embedding model or vector index failure	No — check installation

Intentional degradation (no exception)¶

Situation	`query_type`	Behaviour
Vector search returns no results	`"vector_search"`	Synthesises from training knowledge; `sources` is empty
Similarity below `CONTEXT_CONFIDENCE_THRESHOLD`	`"training_only_response"`	Pack context not injected; Claude answers from training knowledge
Cross-encoder fails to load	N/A	Passthrough — results returned without `ce_score`
Multi-query Haiku call fails	N/A	Proceeds with original query only
All sections filtered by quality scoring	N/A	Falls back to raw article content

Minimal caller pattern¶

from anthropic import APIConnectionError, APIStatusError, APITimeoutError

try:
    result = agent.query(question)
    return result["answer"]
except (APIConnectionError, APITimeoutError):
    return "The AI service is temporarily unavailable. Please try again."
except APIStatusError as e:
    if e.status_code == 401:
        raise RuntimeError("Invalid ANTHROPIC_API_KEY") from e
    raise

See Handle Exceptions for more patterns and Exception Types Reference for the full list.

KG Agent API Reference¶

KnowledgeGraphAgent¶

Constructor¶

Parameters¶

Example¶

query()¶

Parameters¶

Returns¶

query_type Values¶

Example¶

Class Constants¶

Enhancement Module Classes¶

GraphReranker¶

MultiDocSynthesizer¶

FewShotManager¶

CrossEncoderReranker¶

Token Usage Tracking¶

Error Handling¶

Exceptions raised by query()¶

Intentional degradation (no exception)¶

Minimal caller pattern¶

Exceptions raised by `query()`¶