GyanSpark

Send student chat exchanges to the GyanBeej knowledge graph backend, and get back a ready-to-inject summary of what's known about that student — their misconceptions, strengths, and strategies — to ground your own LLM's next reply.

GyanSpark never touches your LLM calls. You call OpenAI, Anthropic, Gemini, or whatever you like yourself, however you like. This is a thin, explicit client for two backend calls, not a proxy or wrapper around your LLM client.

from gyanspark import GyanSpark

gs = GyanSpark()  # reads GYANSPARK_API_KEY from the environment

turn = gs.fill_kg(chat_history, "Which one is cosine again?", roll_number="A-101", name="Jane Doe")

print(turn.context.context_block if turn.context else "(new student, nothing recorded yet)")

fill_kg() resolves the student, fires the actual knowledge-graph write in the background, and — for a student who already exists — hands back a ready-to-inject context summary in that same call, no separate read call needed.

Contents

• Install
• Quick start
• How fill_kg works
• API
  • fill_kg
  • check_call
  • fill_kg_async
  • get_context
• Configuration
• Environment variables
• Error handling
• Testing this repo
• License

Install

pip install gyanspark

Requires Python 3.9+.

Quick start

from gyanspark import GyanSpark

# Reads the key from GYANSPARK_API_KEY if not passed explicitly.
# Construct once per process and reuse across every student.
gs = GyanSpark()

# You make your own LLM call, however you like:
history = [{"role": "user", "content": "I keep mixing up sine and cosine."}]
ai_reply = call_your_llm(history)  # your own code
history.append({"role": "assistant", "content": ai_reply})

# One call: resolves identity, fires the KG write in the background, and —
# for a student who already exists — returns a context summary right away.
turn = gs.fill_kg(history, "Which one is cosine again?", roll_number="A-101", name="Jane Doe")

student_id = turn.student_id  # capture this — reuse it on future calls
if turn.context is not None:
    print(turn.context.context_block)  # ready to paste into your next LLM prompt
else:
    print("brand-new student — nothing recorded yet to summarize")

# Optional: check on the background write later (never block on this).
status = gs.check_call(turn.call_id)

See examples/basic_usage.py for a complete runnable version.

How fill_kg works

fill_kg does three things in one call:

1. Resolves identity synchronously — fast (Redis-cached, no LLM call): student_id if you have it, else roll_number (+ name to auto-create if the student doesn't exist yet).
2. Fires the actual knowledge-graph write in the background and returns immediately — the real classification pipeline (2 sequential Gemini calls) takes 7–45s, so this never blocks the caller. Use check_call to look up what happened to it later; you don't have to.
3. For a student who already exists, calls get_context() synchronously and returns it as turn.context — ready to use for this reply. A brand-new student has nothing recorded yet, so this step is skipped: turn.context is None and turn.is_new_student is True.

turn = gs.fill_kg(history, next_query, student_id=known_id)
# turn.context is ready to use right now for this reply.
# The write is already running in the background — nothing to wait on.

# Whenever convenient (e.g. on the student's next message), check it:
status = gs.check_call(turn.call_id)
if status.status == "failed":
    log.warning("fill_kg write failed: %s", status.error)

Even if you never call check_call, a failed write is still logged automatically (via the SDK's own logger) — so it's never silent, just non-blocking.

API

student_id / roll_number are per-call arguments, not constructor arguments — build one GyanSpark() instance per process and share it across every student your server handles concurrently.

gs.fill_kg(chat_history, query, student_id=None, roll_number=None, name=None, prev_ai="", recent_prompt_timestamp=None, top_n=15, directive=None) -> TurnResult

• chat_history — the exchange that just completed (your student's prior prompt + your AI's reply), written to the graph in the background.
• query — the student's new, not-yet-answered message; what the synchronous context summary (for existing students) is fetched for.
• At least one of student_id / roll_number is required, same as before. prev_ai / directive are passed straight through to the internal get_context() call.
• Returns a TurnResult:
  • student_id, is_new_student — capture student_id for future calls.
  • context — a ContextResult for an existing student, None for a brand-new one (nothing recorded yet to summarize).
  • call_id — pass to check_call() to look up the background write's outcome whenever you want.
• Raises if the identity-resolve step (step 1 above) fails — this is synchronous and blocks the rest of the call, so it raises the same way the old blocking write used to:
  • ValidationError — bad input, caught before any network call.
  • AuthenticationError — invalid/deactivated API key (401/403).
  • GyanSparkTimeoutError / GyanSparkConnectionError — network failure.
  • GyanSparkAPIError — any other non-2xx response.
  • The background write itself never raises here — see check_call.

gs.check_call(call_id) -> CallResult

Looks up a background write kicked off by fill_kg() (or a Future from fill_kg_async(), tracked the same way). Returns CallResult with status — "pending", "succeeded" (result set, a FillKGResult with the same status/summaries shape the old blocking write returned), or "failed" (error set, same exceptions the old blocking write would have raised — auth failures, timeouts, the backend's own pipeline errors, etc.).

Raises ValidationError for an unrecognized call_id.

Caveat: this registry lives in the calling process's memory only — a call_id isn't checkable from a different worker process/replica. Fine for a single-process deployment; don't rely on it across multiple workers.

gs.fill_kg_async(...) -> concurrent.futures.Future[FillKGResult]

Lower-level primitive: fires the raw knowledge-graph write in the background and returns its Future directly — no identity-resolve step, no context fetch, for callers who want manual control instead of fill_kg()'s all-in-one orchestration. Same student_id/roll_number semantics as before (roll_number still auto-creates a student server-side). Exceptions are stored on the Future and only raised when you call .result().

gs.get_context(student_id, query, prev_ai="", directive=None) -> ContextResult

Calls POST /user-context/statements. Built for the hot path.

• Returns a ContextResult with context_block (a string ready to paste into your next LLM prompt, "" if nothing's available), ok (whether the call actually succeeded), and error (populated when ok=False).
• Never raises for a backend/network failure — a timeout, connection error, or non-2xx response degrades to ok=False with an empty context_block.
• Still raises ValidationError for bad input and AuthenticationError for a bad API key — a silently-empty context forever because of a misconfigured key would be a worse failure mode than a loud one.

Configuration

gs = GyanSpark(
    api_key=None,           # falls back to GYANSPARK_API_KEY env var
    base_url=None,          # falls back to GYANSPARK_BASE_URL env var, then https://api.gyanspark.com
    fill_kg_timeout=45,     # seconds — real KG analysis (2 sequential Gemini calls + writes)
    context_timeout=5,      # seconds — backend's <1-2s SLA plus margin
)

Environment variables

See .env.example.

Error handling

fill_kg()'s identity-resolve step raises synchronously — it blocks the rest of the call, so a failure there needs to be loud immediately. The background write never raises to the caller; check check_call() if you want to know its outcome (a failure is also always logged automatically, even if you never check). The read path degrades gracefully because missing context is a soft failure — the student's conversation shouldn't break over it. Auth failures are the one case that's never silent anywhere in this table, since a bad key degrading silently would go unnoticed indefinitely.

Testing this repo

From a checkout of the source repository:

pip install -e ".[dev]"
pytest

Unit tests mock the HTTP layer (requests_mock) — no live backend needed. For end-to-end testing against a real knowledge-graph-gb instance, see smoke_test_AZOPAI.py / smoke_test.py / smoke_test_claude.py (each needs its own LLM provider key plus a real GYANSPARK_API_KEY).

License

MIT — see LICENSE.