LangChain Quickstart

npm install @visiq/harness "langchain@^0.3" "@langchain/openai@^0.3" "@langchain/core@^0.3" "zod@^3"

import { visiq } from "@visiq/harness";
import { AgentExecutor, createOpenAIToolsAgent } from "langchain/agents";
import { createRetrieverTool } from "langchain/tools/retriever";
import { MemoryVectorStore } from "langchain/vectorstores/memory";
import { ChatOpenAI, OpenAIEmbeddings } from "@langchain/openai";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { DynamicTool } from "@langchain/core/tools";

// A small in-memory knowledge base so this snippet runs as pasted.
const vectorStore = await MemoryVectorStore.fromTexts(
  ["Q3 revenue was $4.2M.", "Refunds over $500 require manager approval."],
  [{ classification: "internal" }, { classification: "public" }],
  new OpenAIEmbeddings(),
);
const knowledgeRetriever = vectorStore.asRetriever();

// Your tools — unchanged.
const tools = [
  new DynamicTool({
    name: "issue_refund",
    description: "Issue a refund to a customer",
    func: async (input: string) => {
      const { customerId, amount } = JSON.parse(input);
      return `Refunded $${amount} to ${customerId}`;
    },
  }),
  createRetrieverTool(knowledgeRetriever, {
    name: "search_knowledge",
    description: "Search the company knowledge base",
  }),
];

const prompt = ChatPromptTemplate.fromMessages([
  ["system", "You are a helpful assistant."],
  ["human", "{input}"],
  ["placeholder", "{agent_scratchpad}"],
]);
const llm = new ChatOpenAI({ model: "gpt-4o", temperature: 0 });
const agent = await createOpenAIToolsAgent({ llm, tools, prompt });

// ── One call — governance + audit trail activate here ──
const executor = visiq(new AgentExecutor({ agent, tools }), { agentId: "support-bot" });

const result = await executor.invoke({ input: "What was Q3 revenue?" });

pip install visiq "langchain>=0.3,<0.4" "langchain-openai>=0.2,<0.3" "langchain-core>=0.3,<0.4"

import functools
import inspect

from dotenv import load_dotenv          # pip install python-dotenv
from langchain_core.tools import StructuredTool
from langchain.agents import AgentExecutor, create_openai_tools_agent

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 LangChain 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 = [
    StructuredTool.from_function(func=governed("issue_refund", issue_refund),
                                 name="issue_refund", description="Issue a refund to a customer"),
    StructuredTool.from_function(func=search_knowledge, name="search_knowledge",
                                 description="Search the company knowledge base"),
]

# Report the tool surface once (drives blast-radius inference), then run the agent.
gov.start(tools=[{"name": t.name, "description": t.description} for t in tools])
# … build create_openai_tools_agent(llm, tools, prompt) + AgentExecutor as usual …
gov.flush()  # deliver any buffered audit events before a short-lived process exits