PyWrapAI-Graph

GraphLLM — constructor

graph = GraphLLM(
    llm         = llm,
    enable_system_prompt_versioning = True,
    prompt_file = "support-agent-v1.txt",
    retriever   = my_retriever,   # optional — enables RAG
    rag_k       = 4,
)

If versioning is on but prompt_file is missing, raises ValueError at construction. If retriever is None, all RAG parameters are simply ignored.

Core methods

response = graph.chat("What is the capital of France?")

if response.retrieved_chunks:
    for sc in response.retrieved_chunks:
        print(f"[{sc.score:.3f}] {sc.chunk.text[:80]}")

for chunk in graph.stream(input, history=history):
    print(chunk, end="", flush=True)

result = graph.chat_structured(input, schema=MyModel, history=history)

response = await graph.achat("Hello", history=history)
async for chunk in graph.astream("Write a story", history=history):
    print(chunk, end="")

Proxy properties

with_system_prompt()

Returns a shallow copy with a new system prompt — shares the same underlying LLM (same cache, same TokenTracker). The correct pattern for per-user or per-session prompts without mutating a shared graph.

base_graph    = GraphLLM(llm=llm)
finance_graph = base_graph.with_system_prompt("You are a finance expert.")
support_graph = base_graph.with_system_prompt("You are a support agent.")

# All three share the same cache and TokenTracker
r1 = finance_graph.chat("What is the P/E ratio?")
r2 = support_graph.chat("How do I reset my password?")

PromptRegistry

Used internally by GraphLLM. Files are plain .txt, recommended naming {role}-v{version}.txt for easy A/B testing.

registry  = PromptRegistry(prompts_dir="/abs/path/to/system_prompts_registry")
text      = registry.load("support-agent-v1.txt")   # raises FileNotFoundError if missing
available = registry.list_files()                    # sorted list, [] if dir missing

Tool system

@tool reads a function's type hints and docstring at decoration time and builds the JSON Schema automatically — no hand-written schemas.

@tool
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city."""
    return f"16°C, cloudy in {city}"

# city  → required (no default)
# units → not required (has a default)
get_weather(city="London")   # still callable directly

Type hint → JSON Schema

ToolRegistry

Only the provider files know about wire-format differences — OpenAI wraps in {"type":"function",...}, Anthropic uses input_schema, Gemini builds FunctionDeclaration objects. Your code only ever calls to_schema().

RAG pipeline

Five composable primitives, each replaceable independently: Document → Chunker → embed → BaseVectorStore → VectorRetriever → GraphLLM(retriever=...).

Data types

Chunker

chunker = Chunker(chunk_size=500, overlap=50)
chunks: list[Chunk] = chunker.split(doc)

Split priority: paragraph boundaries → sentence boundaries → hard character cut as a last resort. overlap preserves cross-boundary context between adjacent chunks.

Vector store backends

All six implement BaseVectorStore — swap any one for another with no other code changes. Full constructor options for each are on the Vector Stores page.

VectorRetriever

retriever = VectorRetriever(embedder=embedder, store=store, k=4, min_score=0.0)

n_chunks = retriever.add_documents(docs, chunker=chunker)   # chunk → embed → store, one batched call
results  = retriever.retrieve("What is the capital of France?", k=3)

for sc in results:
    print(f"[{sc.score:.3f}] {sc.chunk.text}")

retrieval_tool()

t = retrieval_tool(retriever=retriever, name="search_knowledge_base",
                    description="Search the knowledge base for relevant facts.")
registry.register(t)

# The LLM can now call "search_knowledge_base" as a tool — the agent
# decides when to retrieve, rather than every call retrieving automatically.

RAG context injection (D4 template)

Injected into the current message text, not a separate message — message count stays constant. Only the original query is ever persisted to history; the <context> block is stripped first, so follow-up turns always retrieve fresh context instead of stale chunks.

Agent — the ReAct loop

Calls the LLM with tool definitions, executes what it requests, appends results, and repeats until a plain-text answer or a limit is hit.

AgentConfig

tool_error_policy

ToolNotFoundError always propagates regardless of policy. ToolArgumentError (bad arguments — validated before the function is even called) falls inside the policy, so the model can self-correct.

Result types

Running the loop

config = AgentConfig(max_turns=8, max_cost_usd=0.10)
agent  = Agent(graph=graph, registry=registry, config=config)

try:
    result = agent.run("What's the weather in London?")
    print(result.content, result.total_cost_usd)
except AgentMaxTurnsError as exc:
    print(exc.partial.total_cost_usd)   # cost up to the limit

result = await agent.arun("Weather in London and Paris?")
print(result.total_cost_usd)