Deploy LangChain agents inside
confidential VMs.

# Standard LangChain stack, no fork, no patch
pip install langchain langchain-openai langgraph langchain-community
pip install pgvector psycopg[binary]   # optional, for confidential RAG

# 5-minute confidential LangChain quickstart
# Swap two lines: base_url + api_key. Everything else is stock LangChain.
# Requests pass through our relay, which sees them in clear (TLS ends at our host).
# To keep plaintext off our servers, run your own inference server in a
# Confidential VM and terminate TLS inside its trust domain.
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 1. Point the OpenAI-compatible client at the confidential endpoint.
llm = ChatOpenAI(
    model="google/gemma-4-31B-turbo-TEE",
    base_url="https://api.voltagegpu.com/v1",
    api_key=os.environ["VOLTAGE_API_KEY"],   # vgpu_... from voltagegpu.com/api-keys
    temperature=0.2,
    max_tokens=2048,
    timeout=60,
)

# 2. Use the chat model exactly like any other ChatOpenAI instance.
messages = [
    SystemMessage(content="You are a privacy-first legal assistant for a French law firm."),
    HumanMessage(content="Summarize the auto-renewal terms in this NDA: ..."),
]

reply = llm.invoke(messages)
print(reply.content)

# 3. Streaming works with .stream() and .astream() exactly like OpenAI.
for chunk in llm.stream(messages):
    print(chunk.content, end="", flush=True)

// LangChain.js / @langchain/openai on our confidential endpoint.
// Requests pass through our relay, which sees them in clear (TLS ends at our host).
// To keep plaintext off our servers, run your own inference server in a
// Confidential VM and terminate TLS inside its trust domain.
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, SystemMessage } from "@langchain/core/messages";

const llm = new ChatOpenAI({
  model: "google/gemma-4-31B-turbo-TEE",
  apiKey: process.env.VOLTAGE_API_KEY,        // vgpu_...
  configuration: {
    baseURL: "https://api.voltagegpu.com/v1", // sovereign endpoint
  },
  temperature: 0.2,
  maxTokens: 2048,
});

const reply = await llm.invoke([
  new SystemMessage("You are a sovereign AI assistant."),
  new HumanMessage("Draft a redline for this auto-renewal clause..."),
]);

console.log(reply.content);

// Streaming
const stream = await llm.stream([
  new HumanMessage("Outline the GDPR Article 28 obligations for processors."),
]);
for await (const chunk of stream) process.stdout.write(chunk.content as string);

# LangGraph state graph, every node uses the confidential endpoint
from typing import TypedDict, List
from langchain_openai import ChatOpenAI
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

llm = ChatOpenAI(
    model="google/gemma-4-31B-turbo-TEE",
    base_url="https://api.voltagegpu.com/v1",
    api_key="vgpu_...",
    temperature=0.1,
)

class GraphState(TypedDict):
    question: str
    plan: str
    context: List[str]
    draft: str
    critique: str
    final: str

# --- Nodes ---

def planner(state: GraphState) -> GraphState:
    msg = llm.invoke([
        SystemMessage(content="Decompose the user question into a 3-step research plan."),
        HumanMessage(content=state["question"]),
    ])
    return {"plan": msg.content}

def researcher(state: GraphState) -> GraphState:
    # Plug in a private retriever here (PGVector / Qdrant on-prem).
    # For brevity we stub the retrieval step.
    return {"context": ["...private firm context retrieved on-prem..."]}

def writer(state: GraphState) -> GraphState:
    msg = llm.invoke([
        SystemMessage(content="Write a draft answer using only the provided context."),
        HumanMessage(content=f"Plan:\n{state['plan']}\n\nContext:\n{state['context']}"
                             f"\n\nQuestion: {state['question']}"),
    ])
    return {"draft": msg.content}

def critic(state: GraphState) -> GraphState:
    msg = llm.invoke([
        SystemMessage(content="Critique the draft for hallucination and missing citations."),
        HumanMessage(content=state["draft"]),
    ])
    return {"critique": msg.content}

def finalizer(state: GraphState) -> GraphState:
    msg = llm.invoke([
        SystemMessage(content="Produce the final answer, addressing the critique."),
        HumanMessage(content=f"Draft:\n{state['draft']}\n\nCritique:\n{state['critique']}"),
    ])
    return {"final": msg.content}

# --- Graph wiring ---
g = StateGraph(GraphState)
g.add_node("plan", planner)
g.add_node("research", researcher)
g.add_node("write", writer)
g.add_node("critique", critic)
g.add_node("finalize", finalizer)
g.add_edge(START, "plan")
g.add_edge("plan", "research")
g.add_edge("research", "write")
g.add_edge("write", "critique")
g.add_edge("critique", "finalize")
g.add_edge("finalize", END)

graph = g.compile(checkpointer=MemorySaver())

result = graph.invoke(
    {"question": "Draft a memo on cross-border data transfers under SCCs 2021/914."},
    config={"configurable": {"thread_id": "matter-2026-0419"}},
)
print(result["final"])

# Tool calling, agent reasons in TDX, tools execute on your machine
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_clauses(matter_id: str, query: str) -> str:
    """Search the firm's internal clause library for a given matter."""
    # Hits an internal API on your network, never leaves your VPC.
    return internal_clause_api(matter_id, query)

@tool
def jurisdiction_check(country: str, regulation: str) -> str:
    """Check whether a regulation applies in a given EU jurisdiction."""
    return internal_jurisdiction_db(country, regulation)

llm = ChatOpenAI(
    model="google/gemma-4-31B-turbo-TEE",
    base_url="https://api.voltagegpu.com/v1",
    api_key="vgpu_...",
    temperature=0,
)

agent = create_react_agent(llm, [search_clauses, jurisdiction_check])

response = agent.invoke({
    "messages": [
        ("user",
         "For matter M-2026-0419, find all auto-renewal clauses and check whether "
         "the German implementation of EU Directive 2019/770 caps them.")
    ]
})
print(response["messages"][-1].content)

# Server-Sent Events streaming, works the same as OpenAI
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="google/gemma-4-31B-turbo-TEE",
    base_url="https://api.voltagegpu.com/v1",
    api_key="vgpu_...",
    streaming=True,
)

# Sync streaming
for chunk in llm.stream([HumanMessage(content="Explain the NIS2 incident-reporting timelines.")]):
    print(chunk.content, end="", flush=True)

# Async streaming for FastAPI / Starlette / Next.js Edge handlers
async def stream_response(question: str):
    async for chunk in llm.astream([HumanMessage(content=question)]):
        yield chunk.content

# LCEL RAG over private documents: local embeddings, LLM in Intel TDX
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_postgres import PGVector

# Confidential LLM (model in Intel TDX)
llm = ChatOpenAI(
    model="Qwen/Qwen3.5-397B-A17B-TEE",
    base_url="https://api.voltagegpu.com/v1",
    api_key="vgpu_...",
    temperature=0,
)

# Embeddings run on your own machine: no confidential embedding model is offered yet.
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")

# Self-hosted vector store inside your VPC, never touches a US SaaS.
vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="firm_matter_notes",
    connection="postgresql+psycopg://app:secret@db.internal:5432/rag",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 6})

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer strictly from the retrieved firm matter notes. "
               "If the answer is not in the context, say so."),
    ("user", "Context:\n{context}\n\nQuestion: {question}"),
])

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke(
    "What is the latest position our firm took on auto-renewal caps in EU SaaS MSAs?"
)
print(answer)

# EU AI Act Article 12, structured logging hook for LangChain
from langchain_core.callbacks import BaseCallbackHandler
from datetime import datetime, timezone
import hashlib
import json

class ArticleTwelveLogger(BaseCallbackHandler):
    """Writes Article 12 compatible event logs to your audit pipeline."""

    def __init__(self, sink):
        self.sink = sink   # e.g. a write-only log bucket, OpenSearch, or Loki

    def _hash(self, text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def on_llm_start(self, serialized, prompts, **kwargs):
        for p in prompts:
            self.sink.write(json.dumps({
                "ts": datetime.now(timezone.utc).isoformat(),
                "event": "llm_start",
                "model": serialized.get("kwargs", {}).get("model"),
                "prompt_sha256": self._hash(p),
                "prompt_chars": len(p),
                "run_id": str(kwargs.get("run_id")),
            }) + "\n")

    def on_llm_end(self, response, **kwargs):
        for gen in response.generations:
            for g in gen:
                self.sink.write(json.dumps({
                    "ts": datetime.now(timezone.utc).isoformat(),
                    "event": "llm_end",
                    "output_sha256": self._hash(g.text),
                    "output_chars": len(g.text),
                    "run_id": str(kwargs.get("run_id")),
                }) + "\n")

    def on_llm_error(self, error, **kwargs):
        self.sink.write(json.dumps({
            "ts": datetime.now(timezone.utc).isoformat(),
            "event": "llm_error",
            "error": str(error),
            "run_id": str(kwargs.get("run_id")),
        }) + "\n")

# Attach the logger to any LangChain invocation.
reply = llm.invoke(messages, config={"callbacks": [ArticleTwelveLogger(audit_sink)]})

Pay per token from your balance, no subscription. Every -TEE model is on the pricing page.