Cookbook¶

import asyncio

from pydantic import BaseModel

from kavalai.agent_service import AgentService
from kavalai.db import db_manager
from kavalai.workflow import WorkflowBuilder


class Message(BaseModel):
    user_message: str


class Reply(BaseModel):
    agent_response: str


async def main():
    await db_manager.init_sqlite()

    bot = (
        WorkflowBuilder("Village guide", llm_model="openai/gpt-5.6-luna")
        .data_model("input", Message)
        .data_model("output", Reply)
        .start("reply")
        .llm("reply", prompt="You are a warm, concise guide to Green Village.",
             inputs={"message": "input"}, output="output", next="end")
        .end()
        .build_engine(
            agent_service=AgentService(
                db_manager.get_sqlite_sessionmaker()
            )
        )
    )

    turns = [
        "I'm Agnes, visiting on Friday.",
        "What did I say my name was?",
    ]
    for turn in turns:
        state = await bot.run({"user_message": turn}, external_id="villager-42")
        print(f"> {turn}\n  {state.output_data['agent_response']}")


asyncio.run(main())

import asyncio

from pydantic import BaseModel

from kavalai import pythontool
from kavalai.agent_service import AgentService
from kavalai.db import db_manager
from kavalai.rag import SqliteRagService
from kavalai.workflow import WorkflowBuilder

FACTS = [
    "Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).",
    "The village pond, Lake Miller, is 1.2 metres deep at its deepest point.",
    "Green Village's only pub, The Rusty Anchor, has been operating since 1923.",
    "The village library owns 1,847 books and is open on Tuesdays and Fridays.",
]

rag = SqliteRagService(":memory:", model="fastembed/BAAI/bge-small-en-v1.5")


class Passages(BaseModel):
    context: str


@pythontool
async def search_village(question: str) -> Passages:
    """Find the village facts most relevant to a question."""
    hits = await rag.query(question, top_k=3)
    return Passages(context="\n".join(f"- {hit.content}" for hit in hits))


class Question(BaseModel):
    user_message: str


class Answer(BaseModel):
    agent_response: str


async def main():
    await rag.index_batch(
        texts=FACTS,
        metadata_list=[{}] * len(FACTS),
        source_ids=[f"fact-{i}" for i in range(len(FACTS))],
    )
    await db_manager.init_sqlite()

    engine = (
        WorkflowBuilder("Village FAQ", llm_model="openai/gpt-5.6-luna")
        .data_model("input", Question)
        .data_model("passages", Passages)
        .data_model("output", Answer)
        .start("retrieve")
        .function(
            "retrieve",
            tool="python://search_village",
            inputs={"question": "input.user_message"},
            output="passages",
            next="answer",
        )
        .llm(
            "answer",
            prompt=(
                "Answer the villager's question using only these facts:\n"
                "{{ context.passages.context }}\n"
                "If they are not enough, say so."
            ),
            inputs={"question": "input"},
            output="output",
            next="end",
        )
        .end()
        .build_engine(
            agent_service=AgentService(
                db_manager.get_sqlite_sessionmaker()
            )
        )
    )
    engine.kernel.register_python_tool("search_village", search_village)

    state = await engine.run({"user_message": "When is the library open?"})
    print(state.output_data["agent_response"])


asyncio.run(main())

from kavalai.workflow import StatsBridge
from kavalai.workflow.tasklog import MemoryTaskLogger

page = {"page": "opening-hours"}
await rag.index_batch(
    texts=[
        "The library is open on Tuesdays and Fridays.",
        "The Rusty Anchor opens at noon.",
    ],
    metadata_list=[page, page],
)

meter = MemoryTaskLogger()
await rag.replace(
    "default",
    texts=["The library is open on Tuesdays, Fridays and Saturdays."],
    metadata_list=[page],
    match=page,
    stats_receiver=StatsBridge(meter),
)
await meter.flush()

print(await rag.count_entries("default"))
for call in meter.model_calls:
    print(call.call_type, call.model, call.batch_size)

from kavalai import Agent, FunctionKernel, make_client
from kavalai.tools.webtools.crawl4ai import crawl_url, web_search

kernel = FunctionKernel()
kernel.register_python_tool("web.search", web_search)
kernel.register_python_tool("web.crawl", crawl_url)
# Registered on the kernel, but withheld from the agent below.
kernel.register_python_tool("db.delete_customer", delete_customer)

researcher = Agent(
    llm_client=make_client("openai/gpt-5.6-luna"),
    kernel=kernel,
    allowed_tools=["python://web.search", "python://web.crawl"],
)

print(await researcher.prompt(
    "Find out what Kaval.AI does and summarise it in three sentences.",
    max_steps=6,
))

from kavalai import WorkflowEngine
from kavalai.testing import ScriptedLlmClient


async def test_repair_requests_route_to_the_repair_handler():
    model = ScriptedLlmClient(
        [
            {"intent": "repair"},
            {"agent_response": "A technician will visit on Monday."},
        ]
    )
    engine = WorkflowEngine.from_yaml_path(
        "council_desk.yaml", client_factory=model
    )

    state = await engine.run({"user_message": "The streetlight is out."})

    assert state.trace == [
        "start", "classify", "route", "repair_reply", "end"
    ]
    assert state.output_data == {
        "agent_response": "A technician will visit on Monday."
    }
    assert "streetlight" in model.calls[0].prompt

import asyncio

from kavalai import WorkflowEngine
from kavalai.rag import SqliteRagService
from kavalai.testing import ScriptedLlmClient, fake_providers

FACTS = [
    "Green Village has 104 residents.",
    "The village pond, Lake Miller, is 1.2 metres deep.",
    "The bakery opens at seven on weekdays.",
]


async def main():
    model = ScriptedLlmClient(
        [
            {"intent": "permit"},
            {"agent_response": "Apply at the parish office."},
        ]
    )
    with fake_providers(llm=model, name="openai"):
        engine = WorkflowEngine.from_yaml_path("council_desk.yaml")
        state = await engine.run({"user_message": "May I build a shed?"})

    print(state.trace)
    print([call.model for call in model.calls])

    with fake_providers() as fakes:
        index = SqliteRagService(":memory:", model="fake/hashing")
        await index.index_batch(FACTS, [{} for _ in FACTS])
        hits = await index.query("How deep is the pond?", top_k=1)

    print(hits[0].content, round(hits[0].similarity, 3))
    print(fakes.embedding.calls[-1])


asyncio.run(main())

import os

from kavalai import OpenAIClient, make_client, register_llm_provider

register_llm_provider(
    "deepseek",
    OpenAIClient,
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

client = make_client("deepseek/deepseek-chat")

register_llm_provider("mycorp", "mycorp_sdk.client.MyCorpClient")

import time

from sklearn.feature_extraction.text import HashingVectorizer

from kavalai import (
    BaseEmbeddingClient,
    ModelCallStat,
    register_embedding_provider,
)
from kavalai.rag import SqliteRagService

FACTS = [
    "Green Village's oldest resident is Agnes Whitlow "
    "(born 02.06.1929).",
    "Green Village has 104 residents.",
    "The village pond, Lake Miller, is 1.2 metres deep at its "
    "deepest point.",
]


class LexicalEmbeddingClient(BaseEmbeddingClient):
    """Hashed bag-of-words vectors: no download, no key, no network."""

    def __init__(self, model, dimensions=512):
        super().__init__(model)
        self.vectorizer = HashingVectorizer(
            n_features=dimensions, norm="l2", alternate_sign=False
        )

    async def compute_embeddings(self, texts, normalize=False,
                                 normalizer=None, **kwargs):
        started = time.perf_counter()
        vectors = self.vectorizer.transform(texts).toarray().tolist()
        stats = ModelCallStat(
            call_type="embedding",
            model=f"lexical/{self.model}",
            batch_size=len(texts),
            total_tokens=0,
            duration_seconds=time.perf_counter() - started,
        )
        return vectors, stats


register_embedding_provider(
    "lexical", LexicalEmbeddingClient, dimensions=512
)

village_index = SqliteRagService(":memory:", model="lexical/word-hash")
await village_index.index_batch(
    texts=FACTS,
    metadata_list=[{}] * len(FACTS),
    source_ids=[f"fact-{i}" for i in range(len(FACTS))],
)

for hit in await village_index.query("How deep is Lake Miller?", top_k=2):
    print(f"{hit.similarity:.3f}  {hit.content}")

register_rag_service(
    "handbook", SqliteRagService,
    filename="handbook.db", model="fastembed/BAAI/bge-small-en-v1.5",
)

engine = (
    WorkflowBuilder("Village guide", llm_model="openai/gpt-5.6-luna")
    .data_model("input", Message)
    .data_model("output", Reply)
    .start("reply")
    .llm("reply", prompt="You are a warm guide to Green Village.",
         inputs={"message": "input"}, output="output", next="end",
         stream_output=True, stream_delta=True)
    .end()
    .build_engine(agent_service=service)
)

question = {"user_message": "Tell me about the tower."}
async for event in engine.run_stream(question):
    if event.type == "partial" and event.name == "reply":
        print(event.value, end="", flush=True)
    elif event.type == "workflow_failed":
        print("\nrun failed:", event.value)

import asyncio

from pydantic import BaseModel, Field

from kavalai import make_client

NOTES = """
Village council, 14 March. Present: Thomas Cook, Greta Lindqvist, Agnes Whitlow.
Greta will order 200kg of flour before the Turnip Festival. Thomas agreed to
get three quotes for repairing the church bell by 1 April. We rejected the
proposal to widen Cobbler's Path. Agnes will ask the library to open on
Saturdays during the festival week.
"""


class ActionItem(BaseModel):
    owner: str
    task: str
    due: str | None = Field(
        default=None, description="Deadline if one was stated."
    )


class Minutes(BaseModel):
    date: str
    attendees: list[str]
    actions: list[ActionItem]
    decisions: list[str]


async def main():
    client = make_client("openai/gpt-5.6-luna")
    minutes = await client.prompt(
        f"Extract the minutes from these notes:\n{NOTES}", response_model=Minutes
    )

    print("date     :", minutes.date)
    print("attendees:", ", ".join(minutes.attendees))
    for action in minutes.actions:
        due = f" (due {action.due})" if action.due else ""
        print(f"  - {action.owner}: {action.task}{due}")


asyncio.run(main())

from typing import Optional

from pydantic import BaseModel

from kavalai import pythontool


class Attempts(BaseModel):
    count: int


@pythontool
def bump(current: Optional[int] = None) -> Attempts:
    """Increment the revision counter."""
    return Attempts(count=(current or 0) + 1)

engine = WorkflowEngine.from_yaml_path("notice_writer.yaml")
engine.kernel.register_python_tool("bump", bump)

state = await engine.run(
    {"topic": "the bake sale at the village hall, Saturday 18 October, 10 am"}
)
print("trace   :", " → ".join(state.trace))
print("attempts:", state.data["attempts"])
print("approved:", state.data["review"]["approved"])
print("notice  :", state.output_data["agent_response"])

import asyncio

from pydantic import BaseModel

from kavalai.agent_service import AgentService
from kavalai.db import db_manager
from kavalai.workflow import WorkflowBuilder


class Note(BaseModel):
    user_message: str


class Tagged(BaseModel):
    agent_response: str
    topic: str
    urgent: bool


NOTES = [
    "The church bell has been stuck since Tuesday.",
    "May I put a beehive in my front garden?",
    "The pub sign fell down and nearly hit someone.",
    "When does the library open?",
]


def build(service):
    return (
        WorkflowBuilder("Noticeboard triage", llm_model="openai/gpt-5.6-luna")
        .data_model("input", Note)
        .data_model("output", Tagged)
        .start("tag")
        .llm(
            "tag",
            prompt=(
                "Tag this note from the Green Village noticeboard. "
                "topic is one of: repair, permit, other. "
                "urgent is true only if someone could get hurt. "
                "agent_response is a one-line acknowledgement."
            ),
            inputs={"note": "input"},
            output="output",
            next="end",
            use_history=False,
        )
        .end()
        .build_engine(agent_service=service)
    )


async def main():
    await db_manager.init_sqlite()
    service = AgentService(db_manager.get_sqlite_sessionmaker())
    engine = build(service)           # one engine, shared by every run

    async def classify(note: str):
        return await engine.run({"user_message": note})

    states = await asyncio.gather(*map(classify, NOTES))
    for note, state in zip(NOTES, states):
        out = state.output_data
        print(f"{out['topic']:<7} urgent={out['urgent']!s:<5} {note}")


asyncio.run(main())

for state in states:
    print(state.token_usage)

# Run 1 — draft a reply and stop. Nothing is sent.
draft = await engine.run(
    {"user_message": "The church bell has been stuck since Tuesday."},
    external_id="ticket-91",
)
show_to_reviewer(draft.output_data["agent_response"])

# The reviewer decides here. That may take minutes or days; the
# process is free to exit in the meantime.

# Run 2 — same session, so the draft and its context are already in history.
final = await engine.run(
    {"user_message": "Approved, but mention the bell ringer's name."},
    external_id="ticket-91",
)

from uuid import UUID

from kavalai.workflow.tasklog import PostgresTaskLogger

tasklog = PostgresTaskLogger(service)
engine = (
    WorkflowBuilder("Village guide", llm_model="openai/gpt-5.6-luna")
    .data_model("input", Message)
    .data_model("output", Reply)
    .start("reply")
    .llm("reply", prompt="You are a warm, concise guide to Green Village.",
         inputs={"message": "input"}, output="output", next="end")
    .end()
    .build_engine(agent_service=service, task_logger=tasklog)
)

state = await engine.run({"user_message": "Is the pub open on Sundays?"})
print(state.token_usage)

await tasklog.flush()
for call in await service.get_model_call_stats(run_id=UUID(state.run_id)):
    print(call.model, call.total_tokens, f"{call.duration_seconds:.2f}s")