KhepriAI

KhepriAI is a Python toolkit for building agent-based applications and related AI workflows.

It provides components for agents, teams, workflows, tools, memory, retrieval, structured outputs, runtime diagnostics, model provider routing, and small neural-network utilities.

Version

1.0.0

Requirements

Python >= 3.10

Installation from source

From the project directory:

python -m pip install .

The command uses setup.py in the source bundle.

Basic usage

from khepri_ai import Agent, ScriptedModel, tool

@tool
def add(a: int, b: int) -> int:
    return a + b

agent = Agent(
    name="MathAgent",
    model=ScriptedModel([
        '<tool_call>{"name":"add","arguments":{"a":2,"b":3}}</tool_call>',
        "The answer is 5.",
    ]),
    tools=[add],
)

result = agent.run("Add 2 and 3")
print(result.output)

Simple API

from khepri_ai import ask, ScriptedModel

answer = ask("Say hello", model=ScriptedModel(["Hello"]))
print(answer)

Useful helpers:

from khepri_ai import ask, run, make_agent, make_team, make_rag_agent, pipeline

• ask(...) returns final text.
• run(...) returns a RunResult.
• make_agent(...) creates an agent with optional tools, toolkits, memory, and safety options.
• make_team(...) creates a team from names.
• make_rag_agent(...) creates an agent with a local knowledge search tool.
• pipeline(...) creates a Workflow from callables.

Agents

Agent combines a model, tools, memory, guardrails, planning, evaluation, and optional tool approval.

from khepri_ai import Agent

agent = Agent(
    name="Assistant",
    model="echo",
    role="General assistant",
)

result = agent.run("Write a short checklist")
print(result.output)

Important methods:

run(task)
arun(task)
batch(tasks)
abatch(tasks)
run_json(task)
run_schema(task, schema)
stream(task)
as_tool()
describe()

Tools

Any Python function can be converted into a tool.

from khepri_ai import tool

@tool
def weather(city: str) -> str:
    return f"Weather for {city}"

Tool-related components:

Tool
ToolRegistry
Toolkit
math_toolkit
time_toolkit
text_toolkit
filesystem_toolkit
memory_toolkit

Teams

Team runs multiple agents with a selected strategy.

from khepri_ai import Agent, Team, ScriptedModel

a = Agent("A", model=ScriptedModel(["first"]))
b = Agent("B", model=ScriptedModel(["second"]))
team = Team([a, b], strategy="sequential")

result = team.run("Work on the task")
print(result.output)

Supported strategies:

sequential
parallel
debate
review
map_reduce
consensus

Workflows

Sequential workflow:

from khepri_ai import Workflow

workflow = (
    Workflow("Text Pipeline")
    .then("strip", lambda text: text.strip())
    .then("upper", lambda text: text.upper())
)

print(workflow.run(" hello "))

Graph workflow:

from khepri_ai import GraphWorkflow

workflow = (
    GraphWorkflow("Router")
    .add_step("classify", lambda text: "refund" if "money" in text else "answer")
    .add_step("refund", lambda value: "refund route")
    .add_step("answer", lambda value: "answer route")
    .connect("classify", "refund", condition=lambda output: output == "refund")
    .connect("classify", "answer")
)

Parallel workflow:

from khepri_ai import ParallelWorkflow

workflow = ParallelWorkflow({
    "upper": lambda text: text.upper(),
    "length": lambda text: len(text),
})

print(workflow.run("hello"))

Memory

Memory components:

ShortTermMemory
LongTermMemory
CompositeMemory
SQLiteMemory
MemoryRecord

Example:

from khepri_ai import CompositeMemory

memory = CompositeMemory()
memory.save("Project note", {"kind": "note"})
records = memory.recall("Project")

Retrieval

Local retrieval components:

KnowledgeBase
Document
TextSplitter
HashingVectorizer
SearchResult

Example:

from khepri_ai import KnowledgeBase

kb = KnowledgeBase.from_texts([
    "KhepriAI includes agents, tools, workflows, and retrieval."
])

results = kb.search("agents workflows")
search_tool = kb.as_tool()

Persistence helpers:

save(path)
load(path)
from_texts(...)
add_documents(...)
deduplicate()
clear()

Loaders

Document loaders:

TextLoader
JSONLoader
CSVLoader
DirectoryLoader

Structured outputs

JSON parsing:

extract_json
JsonOutputParser
json_instructions

Schema validation:

from khepri_ai import Schema

schema = (
    Schema(name="Person")
    .field("name", str)
    .field("age", int)
)

validated = schema.validate({"name": "Ali", "age": "30"})
print(validated)

Schema types can be declared using string names or Python primitive types:

"string"  or str
"integer" or int
"number"  or float
"boolean" or bool
"array"   or list, tuple, set
"object"  or dict

Unsupported types raise SchemaValidationError.

Guardrails, budgets, and approval

Execution controls:

Budget
BudgetExceeded
ForbiddenTermsGuardrail
RequiredTermsGuardrail
MaxLengthGuardrail
RegexGuardrail
NoHtmlGuardrail
AutoApprovalPolicy
RiskBasedApprovalPolicy
ConsoleApprovalPolicy

Example:

from khepri_ai import Agent, Budget, AutoApprovalPolicy

agent = Agent(
    "SafeAgent",
    model="echo",
    budget=Budget(max_model_calls=4, max_tool_calls=8),
    tool_approval=AutoApprovalPolicy(denied_tools={"delete_file"}),
)

Model providers

Model routing uses provider strings.

Examples:

from khepri_ai import create_model

create_model("echo")
create_model("openai:gpt-4o-mini")
create_model("groq:llama-3.1-70b-versatile")
create_model("openrouter:anthropic/claude-3.5-sonnet")
create_model("deepseek:deepseek-chat")
create_model("mistral:mistral-small-latest")
create_model("ollama:llama3.1")
create_model("lmstudio:local-model")

Supported provider names include:

echo
openai
groq
openrouter
deepseek
mistral
together
fireworks
perplexity
xai
cerebras
nvidia
sambanova
huggingface
gemini
anthropic
azure-openai
ollama
lmstudio
vllm
openai-compatible

Provider aliases include:

grok
pplx
hf
nim
local
lm-studio
custom
api
google
claude
azure

Azure OpenAI requires an explicit deployment name:

create_model("azure-openai:deployment-name")

A custom OpenAI-compatible provider can be registered at runtime:

from khepri_ai import register_openai_compatible_provider, create_model

register_openai_compatible_provider(
    "my-provider",
    default_model="my-model",
    base_url="https://local-provider.invalid/v1",
    api_key_env="MY_PROVIDER_API_KEY",
)

model = create_model("my-provider:my-model")

Model wrappers

RetryModel
FallbackModel
RoutedModel
CachedModel
RateLimitedModel

Example:

from khepri_ai import FallbackModel, RetryModel

model = FallbackModel([
    RetryModel("openai:gpt-4o-mini", attempts=2),
    "groq:llama-3.1-8b-instant",
    "echo",
])

Runtime

KhepriRuntime wires shared events, metrics, memory, tools, agents, teams, and workflows.

from khepri_ai import KhepriRuntime, RuntimeConfig, ScriptedModel

runtime = KhepriRuntime(RuntimeConfig(default_model="echo", safe_mode=True))
runtime.agent("Assistant", model=ScriptedModel(["ready"]))

print(runtime.ask("Assistant", "status"))
print(runtime.snapshot())

Health checks

from khepri_ai import run_health_checks

report = run_health_checks()
print(report.to_markdown())
report.raise_for_errors()

Events and metrics

from khepri_ai import EventBus, MetricsCollector, Agent

bus = EventBus()
metrics = MetricsCollector()
bus.subscribe("*", metrics)

agent = Agent("Assistant", model="echo", events=bus)
agent.run("Say hello")

print(metrics.snapshot().to_markdown())

Prompts

from khepri_ai import PromptTemplate, ChatPromptTemplate

prompt = PromptTemplate("Create a plan for {goal}")
print(prompt.render(goal="a task"))

chat = ChatPromptTemplate.from_messages([
    ("system", "You are {role}."),
    ("user", "Task: {task}"),
])

Neural networks

Components:

Dense
Dropout
LayerNorm
Softmax
NeuralNetwork
NeuralNetworkBuilder
NeuralClassifier
NeuralRegressor
Dataset
StandardScaler
MinMaxScaler
OneHotEncoder
EarlyStopping
ConstantLR
StepDecay
ExponentialDecay
gradient_check
cross_validate

Example:

from khepri_ai import NeuralNetworkBuilder, NeuralClassifier, set_seed

set_seed(42)
network = (
    NeuralNetworkBuilder(2, name="classifier")
    .dense(4, activation="tanh")
    .dense(2)
    .softmax()
    .build()
)

x = [[0, 0], [0, 1], [1, 0], [1, 1]]
y = [[1, 0], [0, 1], [0, 1], [1, 0]]

network.train(x, y, epochs=2000, learning_rate=0.1, loss="cce", optimizer="adam")
classifier = NeuralClassifier(network)
print([classifier.predict(row) for row in x])

Neural text classification:

from khepri_ai import train_text_classifier

classifier = train_text_classifier(
    ["refund money", "need refund", "hello support", "general question"],
    ["refund", "refund", "support", "support"],
    epochs=250,
    seed=42,
)

print(classifier.predict("please refund my money"))

Storage

JSONLStore
SQLiteMemory
SQLiteEventSink
ArtifactStore
Blackboard

Task orchestration and evaluation

TaskQueue
TaskSpec
TaskStatus
Benchmark
Scenario
BenchmarkResult

CLI

After installation:

khepri version
khepri doctor
khepri providers
khepri selftest
khepri run "Say hello" --model echo
khepri init my_app --name "My App" --model echo

Source package contents

This source package includes:

khepri_ai/
pyproject.toml
setup.py
MANIFEST.in
README.md
LICENSE
AUTHORS.md
SUPPORT.md
SECURITY.md
CHANGELOG.md
RELEASE.md
CITATION.cff

Author and contact

Mohamed Ashraf
mohamedashrafaidev@gmail.com