SDKs & integrations

LangChain, LangGraph, LlamaIndex & CrewAI

Crowkis is model-agnostic and framework-agnostic. Two primitives cover every framework: a semantic cache that wraps any model call, and durable agent memory. Here's the exact drop-in for each.

You don't restructure your code to adopt Crowkis. Whatever framework you use, the same two ideas apply: cache any model call so repeated and reworded questions are free, and give the agent memory that survives across sessions. The snippets below are copy-paste ready.

LangChain#

What: a semantic LLM cache. Why: LangChain's built-in cache is exact-match, so a reworded prompt misses and you pay again. How: set Crowkis as the cache once, it mirrors LangChain's own set_llm_cache pattern, so every model call in your app is cached by meaning with no chain changes.

python · langchain

from langchain_core.globals import set_llm_cache
from crowkis.integrations.langchain import CrowkisCache

set_llm_cache(CrowkisCache(tenant="support", ttl=3600))

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
llm.invoke("What is your refund policy?")     # miss → cached
llm.invoke("How long do refunds take?")       # semantic HIT → no model call

LangGraph#

What: durable agent memory as a graph node, plus the same LLM cache. Why: graphs restart and fan out; without memory each run relearns the user and re-pays for context. How: recall before the model call, remember after. Memory is scoped to (agent, user) and survives restarts.

python · langgraph

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from crowkis import CrowkisMemory

llm = ChatOpenAI(model="gpt-4o-mini")

class S(TypedDict):
    user_id: str
    question: str
    context: list
    answer: str

def recall(state: S):
    mem = CrowkisMemory(agent="support", user=state["user_id"])
    return {"context": mem.recall(state["question"], k=5)}

def respond(state: S):
    mem = CrowkisMemory(agent="support", user=state["user_id"])
    prompt = f"Known about this user: {state['context']}\n\nQ: {state['question']}"
    answer = llm.invoke(prompt).content
    mem.extract(f"{state['question']}\n{answer}")   # bank durable facts
    return {"answer": answer}

g = StateGraph(S)
g.add_node("recall", recall); g.add_node("respond", respond)
g.set_entry_point("recall"); g.add_edge("recall", "respond"); g.add_edge("respond", END)
app = g.compile()
print(app.invoke({"user_id": "u_42", "question": "Where do I live again?"})["answer"])

Pair this with set_llm_cache(CrowkisCache(...)) and the model calls inside your graph are cached too, memory and caching from one engine.

LlamaIndex#

What: cache your RAG answers. Why: every LlamaIndex query re-runs retrieval and stuffs pages of context into the prompt, repeated questions pay that big bill again. How: wrap the query with the model-agnostic cache.cached() decorator; a repeated-or-rephrased question skips retrieval and the model entirely.

python · llamaindex

from crowkis import Crowkis
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

cache = Crowkis(tenant="docs")
index = VectorStoreIndex.from_documents(SimpleDirectoryReader("./docs").load_data())
engine = index.as_query_engine()

@cache.cached(ttl=3600)
def answer(question: str) -> str:
    return str(engine.query(question))   # retrieval + model, only runs on a real miss

answer("What is the SLA for enterprise plans?")
answer("enterprise plan uptime guarantee?")   # semantic HIT → no retrieval, no model

CrewAI#

What: shared memory and a cache for a crew of agents. Why: multi-agent crews ask overlapping questions and forget everything between runs. How: give each agent a CrowkisMemory for durable context, and wrap its tools/LLM calls with cache.cached() so the crew's overlap costs one answer, not one per agent.

python · crewai

from crowkis import Crowkis, CrowkisMemory

cache = Crowkis(tenant="research-crew")

# durable, per-agent memory across runs
planner_mem = CrowkisMemory(agent="planner", user="project-x")
planner_mem.remember("Project X targets EU launch in Q3")
planner_mem.recall("when does Project X launch?")

# cache any tool / model call the crew makes, model-agnostic
@cache.cached(ttl=3600)
def web_summarize(prompt: str) -> str:
    return any_model_or_tool(prompt)   # first agent pays; the rest reuse for free

Any other framework#

Because Crowkis caches by wrapping your call, it works with anything that produces text, AutoGen, Haystack, a raw provider SDK, or your own loop. If you can put it in a function, you can cache it.

python · works anywhere

from crowkis import Crowkis
cache = Crowkis(tenant="my-app")

@cache.cached(ttl=3600)
def answer(prompt: str) -> str:
    return whatever_model_you_use(prompt)   # OpenAI, Claude, Llama, a tool, anything

Shell & debugging#

Everything is reachable without code, the binary is the server, REPL, and debugger.

shell

crowkis server --port 6379 --data ./.crow    # start the engine
crowkis cli                                    # open the REPL

> CSET "how do refunds work?" "Refunds take 5-7 business days." EX 3600
OK
> CGET "what's the refund timeline?"           # a paraphrase → hit
"Refunds take 5-7 business days."

crowkis why "can I get my money back?"         # which gate vetoed, with scores
crowkis doctor                                  # config, ports, model, data dir