Documentation

LangGraph Integration

Build autonomous AI agents that search, analyze, and organize files using LangGraph and Aionvision

Client-Side Orchestration + Server-Side Intelligence

LangGraph handles client-side orchestration and LLM decision-making. Aionvision handles server-side file intelligence. Together they let you build agents that autonomously work with hundreds or thousands of files — the LLM observes results and decides which Aionvision operations to run next (search, analyze, organize, synthesize) based on what it finds.

When to Use What

Pipelines — Known workflow, server-side, one HTTP call. See Pipelines.

LangChain Chains / Agents — Fixed client-side workflows with prompt templates, structured output, and streaming. See LangChain Integration.

LangGraph — Use when the LLM should decide the workflow dynamically. Full graph control, custom state, human-in-the-loop, branching, and persistence.

Prerequisites

pip install langgraph langchain-anthropic aion

You'll also need an Aionvision API key (AIONVISION_API_KEY) and an Anthropic API key (ANTHROPIC_API_KEY) set in your environment.

Quick Start — ReAct Agent

A minimal LangGraph agent with three Aionvision tools. The LLM decides which tools to call and in what order based on the user's request.

import asyncio
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from aion import AionVision

client: AionVision = None  # initialized in main()

@tool
async def search_images(query: str, limit: int = 20) -> str:
    """Search images by natural language description. Returns matching image IDs and a summary."""
    result = await client.agent_search.images(query, limit=limit)
    return f"Found {result.count} images. IDs: {result.result_ids[:10]}... Summary: {result.summary}"

@tool
async def search_documents(query: str, limit: int = 10) -> str:
    """Search documents by semantic meaning. Returns matching chunks and a summary."""
    result = await client.agent_search.documents(query, limit=limit)
    return f"Found {result.count} results across {len(result.document_ids)} documents. Summary: {result.summary}"

@tool
async def synthesize_report(
    intent: str,
    image_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
) -> str:
    """Generate a report from images and/or documents."""
    result = await client.agent_operations.synthesize(
        intent, image_ids=image_ids, document_ids=document_ids
    )
    return result.summary

async def main():
    global client
    async with AionVision.from_env() as client:
        agent = create_react_agent(
            model=ChatAnthropic(model="claude-sonnet-4-20250514"),
            tools=[search_images, search_documents, synthesize_report],
            prompt="You are a file intelligence assistant. Use the available tools "
                   "to search, analyze, and report on the user's files.",
        )

        result = await agent.ainvoke({
            "messages": [{
                "role": "user",
                "content": "Find all inspection photos from last month and summarize the findings",
            }]
        })
        print(result["messages"][-1].content)

asyncio.run(main())

Defining Aionvision Tools

Each tool wraps one Aionvision SDK method. The docstring is critical — the LLM uses it to decide when to call each tool.

Search Tools

Wrap client.agent_search methods. Return summaries with IDs so the LLM can reference them in later tool calls.

@tool
async def search_images(
    query: str, limit: int = 20, folder_id: str | None = None
) -> str:
    """Search images by natural language description.
    Use this to find photos, screenshots, or any visual content.
    Returns matching image IDs and a summary of what was found."""
    result = await client.agent_search.images(
        query, limit=limit, folder_id=folder_id
    )
    ids = result.result_ids[:10]
    return (
        f"Found {result.count} images. "
        f"Top IDs: {ids}\n"
        f"Summary: {result.summary}"
    )

@tool
async def search_documents(
    query: str, limit: int = 10, document_types: list[str] | None = None
) -> str:
    """Search documents by semantic meaning.
    Use this to find PDFs, reports, contracts, or any text-based files.
    Returns matching document IDs and a summary."""
    result = await client.agent_search.documents(
        query, limit=limit, document_types=document_types
    )
    return (
        f"Found {result.count} results across "
        f"{len(result.document_ids)} documents.\n"
        f"Document IDs: {result.document_ids[:10]}\n"
        f"Summary: {result.summary}"
    )

Operation Tools

Wrap client.agent_operations methods. These act on file IDs found by the search tools.

@tool
async def synthesize_report(
    intent: str,
    image_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
) -> str:
    """Generate a report from images and/or documents.
    Use this after searching to create summaries, assessments, or analyses."""
    result = await client.agent_operations.synthesize(
        intent, image_ids=image_ids, document_ids=document_ids
    )
    return result.summary

@tool
async def analyze_documents(intent: str, document_ids: list[str]) -> str:
    """Analyze and compare documents in depth.
    Use this to extract specific information, compare terms, or find patterns."""
    result = await client.agent_operations.analyze_documents(
        intent, document_ids=document_ids
    )
    return result.summary

@tool
async def organize_files(
    intent: str,
    image_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
    parent_folder_id: str | None = None,
) -> str:
    """Organize files into folders based on the given intent.
    Use this to sort, categorize, or structure files after searching/analyzing."""
    result = await client.agent_operations.organize(
        intent,
        image_ids=image_ids,
        document_ids=document_ids,
        parent_folder_id=parent_folder_id,
    )
    return result.summary

Pipeline Tool (Advanced)

For known multi-step workflows, wrap the Pipeline builder as a single tool. The LLM can invoke it when the task maps to a fixed sequence.

@tool
async def run_pipeline(description: str) -> str:
    """Run a multi-step pipeline for known workflows.
    Use this when the task clearly maps to: search -> analyze -> organize.
    The server handles data wiring between steps automatically."""
    result = await (
        client.pipeline()
        .search_images(description)
        .analyze("Categorize findings")
        .organize("Sort into folders by category")
        .run()
    )
    return f"Pipeline {'succeeded' if result.success else 'failed'}. " + \
           " | ".join(f"[{s.agent}] {s.summary}" for s in result.steps)

Custom State for File Tracking

Extend the basic message state to track file IDs across tool calls. This lets the LLM reference previously found files without re-searching.

State Schema

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    found_image_ids: list[str]       # accumulated across searches
    found_document_ids: list[str]    # accumulated across searches
    analysis_complete: bool
    report_generated: bool

Extracting IDs from Tool Results

A post-tool node that parses tool results and updates the accumulated file IDs in state.

import re

def extract_ids_node(state: AgentState) -> dict:
    """Parse the latest tool message and accumulate file IDs in state."""
    updates: dict = {}
    last_msg = state["messages"][-1]

    if not hasattr(last_msg, "content"):
        return updates

    content = last_msg.content
    # Extract image IDs from tool output
    img_match = re.search(r"Top IDs: \[([^\]]+)\]", content)
    if img_match:
        new_ids = [id.strip().strip("'") for id in img_match.group(1).split(",")]
        updates["found_image_ids"] = list(set(
            state.get("found_image_ids", []) + new_ids
        ))

    # Extract document IDs from tool output
    doc_match = re.search(r"Document IDs: \[([^\]]+)\]", content)
    if doc_match:
        new_ids = [id.strip().strip("'") for id in doc_match.group(1).split(",")]
        updates["found_document_ids"] = list(set(
            state.get("found_document_ids", []) + new_ids
        ))

    return updates

Building the Graph from Scratch

For full control over the execution flow, build the graph manually instead of using create_react_agent.

Graph Definition

from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode

model = ChatAnthropic(model="claude-sonnet-4-20250514")
tools = [search_images, search_documents, synthesize_report,
         analyze_documents, organize_files]
model_with_tools = model.bind_tools(tools)

async def agent_node(state: AgentState) -> dict:
    """Call the LLM to decide the next action."""
    response = await model_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """Route to tools if the LLM made tool calls, otherwise end."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# Build the graph
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_node("extract_ids", extract_ids_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "extract_ids")
builder.add_edge("extract_ids", "agent")

graph = builder.compile()

Running the Graph

result = await graph.ainvoke({
    "messages": [{"role": "user", "content": "Find inspection photos and generate a report"}],
    "found_image_ids": [],
    "found_document_ids": [],
    "analysis_complete": False,
    "report_generated": False,
})

print(result["messages"][-1].content)
print(f"Images found: {len(result['found_image_ids'])}")
print(f"Documents found: {len(result['found_document_ids'])}")

Human-in-the-Loop

Use LangGraph's interrupt mechanism to pause before destructive operations like organizing or deleting files.

Adding an Approval Step

Compile the graph with a checkpointer and interrupt_before on the tool node. The agent will pause after deciding to organize files, letting you inspect the plan and approve or reject.

from langgraph.checkpoint.memory import InMemorySaver

# Compile with interrupt support
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],  # pause before any tool execution
)

config = {"configurable": {"thread_id": "inspection-review"}}

# First run — agent searches and decides to organize
result = await graph.ainvoke(
    {"messages": [{"role": "user", "content": "Organize all site photos by location"}],
     "found_image_ids": [], "found_document_ids": [],
     "analysis_complete": False, "report_generated": False},
    config=config,
)

# Inspect what the agent wants to do
last_msg = result["messages"][-1]
print(f"Agent wants to call: {last_msg.tool_calls}")

# Approve — resume execution
result = await graph.ainvoke(None, config=config)
print(result["messages"][-1].content)

Complete Example — Site Inspection Agent

A full end-to-end agent that handles: "Review all site inspection data and generate a compliance report". The agent searches photos and documents, cross-references findings, synthesizes a report, and optionally organizes files by status.

import asyncio
from typing import Annotated, TypedDict
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import AnyMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from aion import AionVision

# ── State ────────────────────────────────────────────
class InspectionState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# ── Aionvision client (set in main) ─────────────────
client: AionVision = None

# ── Tools ────────────────────────────────────────────
@tool
async def search_images(query: str, limit: int = 30) -> str:
    """Search inspection photos by description."""
    result = await client.agent_search.images(query, limit=limit)
    return (
        f"Found {result.count} images. "
        f"IDs: {result.result_ids[:15]}\n"
        f"Summary: {result.summary}"
    )

@tool
async def search_documents(query: str, limit: int = 10) -> str:
    """Search inspection reports and compliance documents."""
    result = await client.agent_search.documents(query, limit=limit)
    return (
        f"Found {result.count} results across "
        f"{len(result.document_ids)} documents.\n"
        f"Document IDs: {result.document_ids}\n"
        f"Summary: {result.summary}"
    )

@tool
async def synthesize_report(
    intent: str,
    image_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
) -> str:
    """Generate a compliance or assessment report."""
    result = await client.agent_operations.synthesize(
        intent, image_ids=image_ids, document_ids=document_ids
    )
    return result.summary

@tool
async def organize_files(
    intent: str,
    image_ids: list[str] | None = None,
    document_ids: list[str] | None = None,
) -> str:
    """Organize files into folders by status or category."""
    result = await client.agent_operations.organize(
        intent, image_ids=image_ids, document_ids=document_ids
    )
    return result.summary

# ── Graph ────────────────────────────────────────────
tools = [search_images, search_documents, synthesize_report, organize_files]
model = ChatAnthropic(model="claude-sonnet-4-20250514").bind_tools(tools)

SYSTEM_PROMPT = """You are a site inspection assistant. Your workflow:
1. Search for inspection photos and documents
2. Review what you found — if results seem incomplete, search again with different terms
3. Synthesize a compliance report from the combined findings
4. Optionally organize files by compliance status

Always pass file IDs from search results to downstream tools."""

async def agent_node(state: InspectionState) -> dict:
    messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = await model.ainvoke(messages)
    return {"messages": [response]}

def should_continue(state: InspectionState) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

builder = StateGraph(InspectionState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")
graph = builder.compile()

# ── Run ──────────────────────────────────────────────
async def main():
    global client
    async with AionVision.from_env() as client:
        result = await graph.ainvoke({
            "messages": [{
                "role": "user",
                "content": "Review all site inspection data from the past quarter "
                           "and generate a compliance report. Organize any non-compliant "
                           "items into a separate folder.",
            }],
        })

        # Print the full conversation
        for msg in result["messages"]:
            role = getattr(msg, "type", "unknown")
            if role == "ai" and not msg.tool_calls:
                print(f"\nAssistant: {msg.content}")
            elif role == "tool":
                print(f"\n[Tool: {msg.name}] {msg.content[:200]}...")

asyncio.run(main())

Best Practices

Use Pipelines for Known Flows

If the workflow is fixed (search → analyze → organize), use client.pipeline() directly. LangGraph adds value when the LLM should decide the path dynamically.

Keep Tools Focused

Each tool should do one thing. Let the LLM compose them. Don't build a "do everything" tool — the LLM reasons better with single-purpose tools and clear docstrings.

Track State for Context

Use custom state fields to accumulate file IDs across tool calls. This lets the LLM reference previously found files without re-searching, and enables post-processing nodes.

Add Guardrails

Use interrupt_before for destructive operations. Set recursion_limit on the graph to prevent runaway loops. Log tool calls for auditability.