LlamaIndex.TS Quickstart

npm install @visiq/harness llamaindex @llamaindex/workflow @llamaindex/openai zod

import { visiq } from "@visiq/harness";
import { tool } from "llamaindex";
import { agent } from "@llamaindex/workflow";
import { openai } from "@llamaindex/openai";
import { z } from "zod";

// Any function returning document-shaped results works as a RAG source.
const searchDocs = async (query: string) => [
  { pageContent: `Q3 revenue was $4.2M. (matched: ${query})`, metadata: { classification: "internal" } },
];

// Your tools — unchanged.
const tools = [
  tool({
    name: "issue_refund",
    description: "Issue a refund to a customer",
    parameters: z.object({ customerId: z.string(), amount: z.number() }),
    execute: async ({ customerId, amount }) => `Refunded $${amount} to ${customerId}`,
  }),
  tool({
    name: "search_knowledge",
    description: "Search the company knowledge base",
    parameters: z.object({ query: z.string() }),
    execute: async ({ query }) => searchDocs(query),
  }),
];

// ── One call — governance + per-run session activate here ──
const supportAgent = visiq(
  agent({ name: "support", llm: openai({ model: "gpt-4o" }), tools, systemPrompt: "You are a helpful assistant." }),
  { agentId: "support-bot" },
);

const result = await supportAgent.run("What was Q3 revenue?");
console.log(result.data.result);

pip install visiq "llama-index-core>=0.11,<0.13" "llama-index-llms-openai>=0.2,<0.4"

import asyncio
import functools
import inspect

from dotenv import load_dotenv          # pip install python-dotenv
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

from visiq import Governor, ToolBlocked  # pip install visiq

load_dotenv()                            # VISIQ_API_KEY / VISIQ_AGENT_ID (+ VISIQ_ENDPOINT for onprem)

gov = Governor(agent_id="support-bot")


def governed(name, fn):
    # functools.wraps preserves the real signature so LlamaIndex builds a correct
    # per-parameter schema (a bare **kwargs wrapper would blind arg-matching rules).
    sig = inspect.signature(fn)

    @functools.wraps(fn)
    def tool_fn(*a, **kw):
        bound = sig.bind(*a, **kw)
        bound.apply_defaults()
        args = dict(bound.arguments)
        try:
            # gate_tool decides BEFORE the body runs; the callback receives the
            # EFFECTIVE (redacted-on-mask) arguments, never the originals.
            return gov.gate_tool(name, args, lambda effective: fn(**effective))
        except ToolBlocked as e:
            return f"[BLOCKED BY POLICY] {e.reason}"

    tool_fn.__name__ = name
    return tool_fn


def issue_refund(customer_id: str, amount: int) -> str:
    return f"Refunded ${amount} to {customer_id}"


def scenario_search(query: str) -> list:
    # A stand-in retrieval source — your real RAG store returns the same shape: a
    # list of {"page_content", "metadata"} dicts that gate_documents can filter,
    # drop, or redact before the model sees them.
    return [
        {"page_content": f"Q3 revenue was $4.2M. (matched: {query})",
         "metadata": {"classification": "internal"}},
    ]


def search_knowledge(query: str) -> str:
    # Retrieval governance: drop / redact documents before the model sees them.
    docs = gov.gate_documents(scenario_search(query), query=query)
    return "\n\n".join(d["page_content"] for d in docs)


tools = [
    FunctionTool.from_defaults(fn=governed("issue_refund", issue_refund),
                               name="issue_refund", description="Issue a refund to a customer"),
    FunctionTool.from_defaults(fn=search_knowledge, name="search_knowledge",
                               description="Search the company knowledge base"),
]


async def main():
    # Report the tool surface once (drives blast-radius inference).
    gov.start(tools=[{"name": t.metadata.name, "description": t.metadata.description} for t in tools])
    agent = FunctionAgent(tools=tools, llm=OpenAI(model="gpt-4o"),
                          system_prompt="You are a helpful assistant.")
    print(await agent.run("What was Q3 revenue?"))
    gov.flush()  # deliver any buffered audit events before the process exits


asyncio.run(main())