AI agent architectures, explained simply

import { StateGraph, StateSchema, ReducedValue, START, END } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
import type { AIMessage } from "@langchain/core/messages";
import { z } from "zod";

const TIERS = ["small", "medium", "large"] as const;
type Tier = (typeof TIERS)[number];

const models: Record<Tier, ChatAnthropic> = {
  small: new ChatAnthropic({ model: "claude-haiku-5-5" }),
  medium: new ChatAnthropic({ model: "claude-sonnet-5-5" }),
  large: new ChatAnthropic({ model: "claude-opus-5-5" }),
};
// $ per million tokens, to be filled in from the current price list
declare const PRICING: Record<Tier, { input: number; output: number }>;
declare function estimateUsd(tier: Tier, task: string): number;

type CostLine = { step: string; tier: Tier; inTok: number; outTok: number; usd: number };
const costLine = (step: string, tier: Tier, raw: AIMessage): CostLine => {
  const u = raw.usage_metadata ?? { input_tokens: 0, output_tokens: 0, total_tokens: 0 };
  const p = PRICING[tier];
  const usd = (u.input_tokens * p.input + u.output_tokens * p.output) / 1e6;
  return { step, tier, inTok: u.input_tokens, outTok: u.output_tokens, usd };
};

const State = new StateSchema({
  task: z.string(),
  budgetUsd: z.number(),
  tier: z.enum(TIERS).default("small"),
  answer: z.string().default(""),
  confidence: z.number().default(0),
  // Cost log per decision: each call appends its own line.
  ledger: new ReducedValue(z.array(z.custom<CostLine>()).default(() => []), {
    reducer: (all, added) => all.concat(added),
  }),
});
type S = typeof State.State;
const spent = (s: S) => s.ledger.reduce((t, l) => t + l.usd, 0);

// Routing by complexity: the smallest model does the sorting.
async function triage(s: S) {
  const { raw, parsed } = await models.small
    .withStructuredOutput(z.object({ complexity: z.enum(["simple", "moderate", "hard"]) }), { method: "jsonSchema", includeRaw: true })
    .invoke(`Assess the complexity of this task: ${s.task}`);
  const tier = ({ simple: "small", moderate: "medium", hard: "large" } as const)[parsed.complexity];
  return { tier, ledger: [costLine("triage", "small", raw as AIMessage)] };
}

async function solve(s: S) {
  const { raw, parsed } = await models[s.tier]
    .withStructuredOutput(z.object({ answer: z.string(), confidence: z.number().min(0).max(1) }), { method: "jsonSchema", includeRaw: true })
    .invoke(s.task);
  return { ...parsed, ledger: [costLine("solve", s.tier, raw as AIMessage)] };
}

const escalate = (s: S) => ({ tier: TIERS[TIERS.indexOf(s.tier) + 1] });

// Early exit on confidence, escalation for as long as the budget allows.
function next(s: S) {
  if (s.confidence >= 0.85) return END;
  const up = TIERS[TIERS.indexOf(s.tier) + 1];
  if (!up) return END;
  if (spent(s) + estimateUsd(up, s.task) > s.budgetUsd) return END;
  return "escalate";
}

export const graph = new StateGraph(State)
  .addNode("triage", triage)
  .addNode("solve", solve)
  .addNode("escalate", escalate)
  .addEdge(START, "triage")
  .addEdge("triage", "solve")
  .addConditionalEdges("solve", next, ["escalate", END])
  .addEdge("escalate", "solve")
  .compile();

// The ledger goes into your database: it is what tells you how much the routing saves.
// const out = await graph.invoke({ task, budgetUsd: 0.05 });
// await analytics.insert(out.ledger.map((l) => ({ ...l, taskId })));

import { StateGraph, StateSchema, ReducedValue, Send, START, END } from "@langchain/langgraph";
import type { StructuredToolInterface } from "@langchain/core/tools";
import { z } from "zod";
import { llm } from "./models";

type Capability = "crm.read" | "crm.write" | "billing.read" | "email.send";

// Catalogue: each tool declares the capabilities it requires.
declare const catalog: { tool: StructuredToolInterface; requires: Capability[] }[];

// Permission scoping: only the tools covered by the caller's scopes.
const allowedFor = (scopes: Capability[]) =>
  catalog.filter((e) => e.requires.every((c) => scopes.includes(c))).map((e) => e.tool);

const Step = z.object({
  tool: z.string(),
  args: z.record(z.string(), z.unknown()),
  reason: z.string(),
});
type Step = z.infer<typeof Step>;
type ToolResult = { tool: string; ok: boolean; output: string };

const State = new StateSchema({
  task: z.string(),
  scopes: z.array(z.custom<Capability>()),
  plan: z.array(Step).default(() => []),
  // Parallel results accumulate: without a reducer, the last one would overwrite the others.
  results: new ReducedValue(z.array(z.custom<ToolResult>()).default(() => []), {
    reducer: (all, added) => all.concat(added),
  }),
  final: z.string().default(""),
});
type S = typeof State.State;

// Capability-based routing: the planner only sees the authorised tools.
async function planner(s: S) {
  const allowed = allowedFor(s.scopes);
  const catalog = allowed.map((t) => `- ${t.name}: ${t.description}`).join("\n");
  const { steps } = await llm.withStructuredOutput(z.object({ steps: z.array(Step) }), { method: "jsonSchema" }).invoke(
    `Available tools:\n${catalog}\n\nTask: ${s.task}\n` +
      "List the independent calls needed (they will run in parallel).",
  );
  // Defence in depth: filter again, the model may invent a tool.
  const names = new Set(allowed.map((t) => t.name));
  return { plan: steps.filter((st) => names.has(st.tool)) };
}

// Parallel execution: one Send per step, all in the same super-step.
const fanOut = (s: S) =>
  s.plan.length ? s.plan.map((step) => new Send("execute", { step })) : "resolve";

async function execute({ step }: { step: Step }) {
  const t = catalog.find((e) => e.tool.name === step.tool)!.tool;
  try {
    const output = await t.invoke(step.args);
    return { results: [{ tool: step.tool, ok: true, output: String(output) }] };
  } catch (e) {
    return { results: [{ tool: step.tool, ok: false, output: (e as Error).message }] };
  }
}

// Conflict resolution: an explicit priority rule between sources of truth.
async function resolve(s: S) {
  const res = await llm.invoke(
    `Task: ${s.task}\nTool results:\n${JSON.stringify(s.results, null, 2)}\n\n` +
      "If two sources contradict each other, billing takes precedence over the CRM. Flag each discrepancy explicitly.",
  );
  return { final: res.text };
}

export const graph = new StateGraph(State)
  .addNode("planner", planner)
  .addNode("execute", execute)
  .addNode("resolve", resolve)
  .addEdge(START, "planner")
  .addConditionalEdges("planner", fanOut, ["execute", "resolve"])
  .addEdge("execute", "resolve")
  .addEdge("resolve", END)
  .compile();

// await graph.invoke({ task: "Summarise the ACME account", scopes: ["crm.read", "billing.read"] });

import { StateGraph, StateSchema, MessagesValue, START, END, GraphRecursionError } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { AIMessage, HumanMessage, SystemMessage } from "@langchain/core/messages";
import { z } from "zod";
import { llm, fast } from "./models";

declare function searchInternalDocs(query: string): Promise<string>;

const searchDocs = tool(({ query }) => searchInternalDocs(query), {
  name: "search_docs",
  description: "Searches the internal documentation",
  schema: z.object({ query: z.string() }),
});
const tools = [searchDocs];
const model = llm.bindTools(tools);
// The history contains tool_use blocks: the Anthropic API requires the tools to stay declared.
// tool_choice "none" declares them without allowing them to be called.
const critic = fast.bindTools(tools, { tool_choice: "none" });
const finalizer = llm.bindTools(tools, { tool_choice: "none" });

const MAX_ITERATIONS = 8;
const REFLECT_EVERY = 3;

const State = new StateSchema({
  messages: MessagesValue,
  iterations: z.number().default(0),
});
type S = typeof State.State;

// Reason + Act: the model reasons, then calls a tool (or not).
async function agent(s: S) {
  const res = await model.invoke([
    new SystemMessage("Reason step by step, use the tools if needed, then answer."),
    ...s.messages,
  ]);
  return { messages: [res], iterations: s.iterations + 1 };
}

// Reflect: periodic self-critique to break unproductive loops.
async function reflect(s: S) {
  const critique = await critic.invoke([
    ...s.messages,
    new HumanMessage("In 2 sentences: are you making progress towards the answer or going round in circles? What should change?"),
  ]);
  return { messages: [new HumanMessage(`[self-critique] ${critique.text}`)] };
}

// Graceful degradation: a partial answer rather than an error.
async function degrade(s: S) {
  const res = await finalizer.invoke([
    ...s.messages,
    new HumanMessage("Iteration limit reached. Give the best partial answer and list what is missing."),
  ]);
  return { messages: [res] };
}

// Observe: after the tools, the system's policy decides what happens next.
const afterTools = (s: S) => {
  if (s.iterations >= MAX_ITERATIONS) return "degrade";
  if (s.iterations % REFLECT_EVERY === 0) return "reflect";
  return "agent";
};

export const graph = new StateGraph(State)
  .addNode("agent", agent)
  .addNode("tools", new ToolNode(tools))
  .addNode("reflect", reflect)
  .addNode("degrade", degrade)
  .addEdge(START, "agent")
  .addConditionalEdges("agent", toolsCondition, ["tools", END])
  .addConditionalEdges("tools", afterTools, ["agent", "reflect", "degrade"])
  .addEdge("reflect", "agent")
  .addEdge("degrade", END)
  .compile();

// Technical safety net: recursionLimit, in super-steps.
export async function run(question: string) {
  try {
    return await graph.invoke({ messages: [new HumanMessage(question)] }, { recursionLimit: 40 });
  } catch (e) {
    if (e instanceof GraphRecursionError) return { messages: [new AIMessage("I couldn't reach a conclusion.")] };
    throw e;
  }
}

import { StateGraph, StateSchema, ReducedValue, Command, GraphRecursionError, START, END } from "@langchain/langgraph";
import { createAgent, type ReactAgent } from "langchain";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { llm, fast } from "./models";

// Ports: web search, CRM and billing.
declare function webSearch(query: string): Promise<string>;
declare function crmAccount(company: string): Promise<string>;
declare function listInvoices(company: string): Promise<string>;

const webSearchTool = tool(({ query }) => webSearch(query), {
  name: "web_search",
  description: "Web search",
  schema: z.object({ query: z.string() }),
});
const crmTool = tool(({ company }) => crmAccount(company), {
  name: "crm_account",
  description: "Account record in the CRM",
  schema: z.object({ company: z.string() }),
});
const invoicesTool = tool(({ company }) => listInvoices(company), {
  name: "list_invoices",
  description: "Invoices and payments for the account",
  schema: z.object({ company: z.string() }),
});

// Each specialist is a full ReAct agent (createAgent): its own prompt, tools and context.
const researcherAgent = createAgent({
  model: llm,
  tools: [webSearchTool],
  systemPrompt: "You research a company: news, public tenders, hiring. Return a sourced summary.",
});
const analystAgent = createAgent({
  model: llm,
  tools: [crmTool, invoicesTool],
  systemPrompt: "You analyse the business relationship: contracts, invoices, unpaid bills. Return a summary with figures.",
});

type Report = { agent: string; summary: string };
const MAX_ROUNDS = 6;

const State = new StateSchema({
  brief: z.string(),
  task: z.string().default(""),
  rounds: z.number().default(0),
  // Each specialist appends its report: the supervisor only sees these summaries.
  reports: new ReducedValue(z.array(z.custom<Report>()).default(() => []), {
    reducer: (all, added) => all.concat(added),
  }),
  final: z.string().default(""),
});
type S = typeof State.State;

const Route = z.object({
  next: z.enum(["researcher", "analyst", "writer", "FINISH"]),
  task: z.string().describe("Short instruction for the chosen agent"),
});

// The supervisor picks who steps in and with what instruction; the code bounds the loop.
async function supervisor(s: S) {
  if (s.final || s.rounds >= MAX_ROUNDS) return new Command({ goto: END });
  const done = s.reports.map((r) => `## ${r.agent}\n${r.summary}`).join("\n\n") || "(nothing yet)";
  const route = await fast.withStructuredOutput(Route, { method: "jsonSchema" }).invoke(
    `You coordinate the preparation of a client meeting.\nRequest: ${s.brief}\n\nAlready done:\n${done}\n\n` +
      "Who should step in now, and with what instruction? The writer goes last.",
  );
  if (route.next === "FINISH") return new Command({ goto: END });
  return new Command({ goto: route.next, update: { task: route.task, rounds: s.rounds + 1 } });
}

// Each specialist starts with a blank context: the supervisor's instruction, nothing else.
// A specialist that loops returns a partial report instead of failing the whole graph.
const delegate = (name: string, agent: ReactAgent) => async (s: S) => {
  try {
    const out = await agent.invoke({ messages: [{ role: "user", content: s.task }] }, { recursionLimit: 20 });
    return { reports: [{ agent: name, summary: out.messages.at(-1)?.text ?? "" }] };
  } catch (e) {
    if (!(e instanceof GraphRecursionError)) throw e;
    return { reports: [{ agent: name, summary: "Research cut short: step limit reached." }] };
  }
};

const researcher = delegate("researcher", researcherAgent);
const analyst = delegate("analyst", analystAgent);

// The writer needs no tools: a single model call is enough.
async function writer(s: S) {
  const notes = s.reports.map((r) => `## ${r.agent}\n${r.summary}`).join("\n\n");
  const res = await llm.invoke(`Write a one-page meeting brief from these notes.\n\n${notes}`);
  return { final: res.text };
}

export const graph = new StateGraph(State)
  .addNode("supervisor", supervisor, { ends: ["researcher", "analyst", "writer", END] })
  .addNode("researcher", researcher)
  .addNode("analyst", analyst)
  .addNode("writer", writer)
  .addEdge(START, "supervisor")
  .addEdge("researcher", "supervisor")
  .addEdge("analyst", "supervisor")
  .addEdge("writer", "supervisor")
  .compile();

import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import type { DocumentInterface } from "@langchain/core/documents";
import { z } from "zod";
import { llm } from "./models";

// Ports: vector store (pgvector, Qdrant…) and web search (Tavily, Exa…).
interface Retriever { search(q: string, k: number): Promise<DocumentInterface[]> }
interface WebSearch { search(q: string): Promise<DocumentInterface[]> }
declare const retriever: Retriever;
declare const web: WebSearch;

const Answer = z.object({
  text: z.string(),
  citations: z.array(z.string()).describe("ids of the passages used, e.g. doc-12#3"),
  confidence: z.number().min(0).max(1),
});

const State = new StateSchema({
  question: z.string(),
  docs: z.array(z.custom<DocumentInterface>()).default(() => []),
  answer: Answer.extend({ flagged: z.boolean().optional() }).optional(),
  usedWeb: z.boolean().default(false),
});
type S = typeof State.State;

const retrieve = async (s: S) => ({ docs: await retriever.search(s.question, 6) });
const webSearch = async (s: S) => ({ docs: await web.search(s.question), usedWeb: true });

async function generate(s: S) {
  const context = s.docs.map((d) => `[${d.metadata.id}] ${d.pageContent}`).join("\n\n");
  const res = await llm.withStructuredOutput(Answer, { method: "jsonSchema" }).invoke([
    ["system",
      "Answer ONLY from the context. Every claim cites an id in square brackets. " +
      "If the context is not enough, say so and set confidence < 0.5."],
    ["user", `Context:\n${context}\n\nQuestion: ${s.question}`],
  ]);

  // Deterministic grounding: we don't take the model's word for it.
  const known = new Set(s.docs.map((d) => String(d.metadata.id)));
  const grounded = res.citations.length > 0 && res.citations.every((c) => known.has(c));
  return { answer: { ...res, confidence: grounded ? res.confidence : 0 } };
}

const flagLowConfidence = (s: S) => ({ answer: { ...s.answer!, flagged: true } });

function route(s: S) {
  if (s.answer!.confidence >= 0.6) return END;
  return s.usedWeb ? "flag" : "webSearch";
}

export const graph = new StateGraph(State)
  .addNode("retrieve", retrieve)
  .addNode("generate", generate)
  .addNode("webSearch", webSearch)
  .addNode("flag", flagLowConfidence)
  .addEdge(START, "retrieve")
  .addEdge("retrieve", "generate")
  .addConditionalEdges("generate", route, ["webSearch", "flag", END])
  .addEdge("webSearch", "generate")
  .addEdge("flag", END)
  .compile();

import { StateGraph, StateSchema, ReducedValue, Command, MemorySaver, START, END, interrupt } from "@langchain/langgraph";
import { z } from "zod";
import { llm } from "./models";

type Ticket = { id: string; customer: string; body: string; orderTotal: number };
interface Refunds { apply(ticketId: string, d: Decision, idempotencyKey: string): Promise<void> }
declare const refunds: Refunds;

const Decision = z.object({
  action: z.enum(["refund", "reject", "escalate"]),
  amount: z.number().nonnegative().optional(),
  rationale: z.string(),
  confidence: z.number().min(0).max(1),
});
type Decision = z.infer<typeof Decision>;

type AuditEntry = { at: string; actor: string; event: string; data?: unknown };
const audit = (actor: string, event: string, data?: unknown): AuditEntry => ({
  at: new Date().toISOString(), actor, event, data,
});

type HumanReview = {
  decision: "approve" | "edit" | "reject";
  reviewer: string;
  edited?: Decision;
  comment?: string;
};

const State = new StateSchema({
  ticket: z.custom<Ticket>(),
  proposal: Decision.optional(),
  approved: Decision.nullable().default(null),
  // Audit trail: every event is appended, nothing is ever erased.
  audit: new ReducedValue(z.array(z.custom<AuditEntry>()).default(() => []), {
    reducer: (all, added) => all.concat(added),
  }),
});
type S = typeof State.State;

async function propose(s: S) {
  const proposal = await llm.withStructuredOutput(Decision, { method: "jsonSchema" }).invoke(
    `Customer service ticket (order of €${s.ticket.orderTotal}):\n${s.ticket.body}\nPropose an action and your confidence.`,
  );
  return { proposal, audit: [audit("agent", "proposed", proposal)] };
}

// Only a small, safe refund goes through on its own: everything else (low confidence, large amount,
// missing amount, rejection or escalation) goes to a human.
const needsHuman = (s: S) => {
  const p = s.proposal!;
  return p.action === "refund" && p.amount !== undefined && p.amount <= 200 && p.confidence >= 0.8
    ? "autoApprove"
    : "review";
};

const autoApprove = (s: S) => ({ approved: s.proposal!, audit: [audit("policy", "auto_approved")] });

function review(s: S) {
  // On resume, the node is replayed from the start: no side effects before interrupt().
  const human = interrupt<{ ticketId: string; proposal: Decision }, HumanReview>({
    ticketId: s.ticket.id,
    proposal: s.proposal!,
  });
  const entry = audit(human.reviewer, `human_${human.decision}`, { comment: human.comment, edited: human.edited });

  if (human.decision === "reject") {
    return new Command({ goto: END, update: { approved: null, audit: [entry] } });
  }
  const approved = human.decision === "edit" && human.edited ? human.edited : s.proposal!;
  return new Command({ goto: "execute", update: { approved, audit: [entry] } });
}

async function execute(s: S) {
  await refunds.apply(s.ticket.id, s.approved!, `refund:${s.ticket.id}`);
  return { audit: [audit("system", "executed", s.approved)] };
}

export const graph = new StateGraph(State)
  .addNode("propose", propose)
  .addNode("autoApprove", autoApprove)
  .addNode("review", review, { ends: ["execute", END] })
  .addNode("execute", execute)
  .addEdge(START, "propose")
  .addConditionalEdges("propose", needsHuman, ["review", "autoApprove"])
  .addEdge("autoApprove", "execute")
  .addEdge("execute", END)
  .compile({ checkpointer: new MemorySaver() }); // in production: PostgresSaver

// One thread per ticket: that is what lets you resume later.
const thread = (ticketId: string) => ({ configurable: { thread_id: `ticket:${ticketId}` } });

// When paused, the result contains __interrupt__: the proposal to show the case handler.
export const handleTicket = (ticket: Ticket) => graph.invoke({ ticket }, thread(ticket.id));

// Later, from the back office (another HTTP request):
export const resume = (ticketId: string, review: HumanReview) =>
  graph.invoke(new Command({ resume: review }), thread(ticketId));