Aıon

Documentation

LangChain Integration

Build AI-powered file workflows using LangChain chains, tools, and agents with Aionvision

Chains + Tools + Server-Side Intelligence

LangChain provides prompt templates, output parsers, LCEL chains, and tool calling. Aionvision provides server-side file intelligence — search, analyze, organize, and synthesize. Together they let you build structured workflows, from simple tool-calling chains to full agents, over your file collections.

When to Use What

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

LangChain Chains (LCEL) — Fixed client-side workflow with prompt templates, structured output, and streaming.

LangChain Agent (create_react_agent) — Simple agent with automatic tool loop, minimal boilerplate.

LangGraph — Full graph control, custom state, human-in-the-loop, branching. See LangGraph Integration.

Prerequisites

pip install langchain 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 — Tool Calling

The core pattern: define tools wrapping Aionvision SDK calls, bind them to a model, and manually handle the tool call loop. Simple and transparent — you see every step.

import asyncio
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
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]}\nSummary: {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:
        model = ChatAnthropic(model="claude-sonnet-4-20250514")
        tools = [search_images, synthesize_report]
        model_with_tools = model.bind_tools(tools)
        tools_by_name = {t.name: t for t in tools}


        # Initial request
        messages = [HumanMessage("Find inspection photos and summarize findings")]
        response = await model_with_tools.ainvoke(messages)
        messages.append(response)


        # Manual tool loop
        while response.tool_calls:
            for call in response.tool_calls:
                tool_fn = tools_by_name[call["name"]]
                result = await tool_fn.ainvoke(call["args"])
                messages.append(ToolMessage(content=result, tool_call_id=call["id"]))


            response = await model_with_tools.ainvoke(messages)
            messages.append(response)


        print(response.content)


asyncio.run(main())

Simple Agent with create_react_agent

For most use cases, create_react_agent handles the tool loop automatically. Same tools, less boilerplate.

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


client: AionVision = None


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


@tool
async def synthesize_report(
    intent: str, image_ids: list[str] | None = None
) -> str:
    """Generate a report from images."""
    result = await client.agent_operations.synthesize(intent, image_ids=image_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, synthesize_report],
            prompt="You are a file intelligence assistant. Search files and generate "
                   "reports based on user requests.",
        )


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

When to choose what

Use create_react_agent for most cases. Use the manual tool loop (above) when you need fine-grained control over each step. Use LangGraph when you need custom state, human-in-the-loop, or complex branching.

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.

With @tool Decorator (Recommended)

The simplest way to define tools. Works with both search and operation methods.

from langchain_core.tools import tool


# ── Search Tools ──────────────────────────────────────


@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 ───────────────────────────────────


@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

With BaseTool Subclass (Stateful Tools)

Use BaseTool when you need the tool to hold state, like a client reference as an instance attribute.

from pydantic import ConfigDict
from langchain_core.tools import BaseTool
from aion import AionVision


class AionSearchTool(BaseTool):
    name: str = "search_images"
    description: str = (
        "Search images by natural language description. "
        "Returns matching image IDs and a summary."
    )
    client: AionVision


    model_config = ConfigDict(arbitrary_types_allowed=True)


    def _run(self, query: str, limit: int = 20) -> str:
        raise NotImplementedError("Use async")


    async def _arun(self, query: str, limit: int = 20) -> str:
        result = await self.client.agent_search.images(query, limit=limit)
        return (
            f"Found {result.count} images. "
            f"IDs: {result.result_ids[:10]}\n"
            f"Summary: {result.summary}"
        )


# Usage — client must be initialized inside an async context manager
async with AionVision.from_env() as client:
    search_tool = AionSearchTool(client=client)
    # Use search_tool with your agent/chain inside this block

LCEL Chains with Aionvision

Use LangChain Expression Language (LCEL) to compose Aionvision calls into deterministic, streamable chains. No agent loop — each step runs in order.

Pattern 1: Search → LLM Summary

Search files with Aionvision, then pipe results into an LLM for summarization.

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda


async def search_step(query: str) -> dict:
    result = await client.agent_search.images(query, limit=20)
    return {
        "query": query,
        "count": result.count,
        "summary": result.summary,
        "ids": result.result_ids[:10],
    }


prompt = ChatPromptTemplate.from_template(
    "Based on the search results below, provide a concise summary.\n\n"
    "Query: {query}\n"
    "Found: {count} images\n"
    "Results: {summary}"
)


model = ChatAnthropic(model="claude-sonnet-4-20250514")
chain = RunnableLambda(search_step) | prompt | model | StrOutputParser()


# Run
summary = await chain.ainvoke("inspection photos from site A")

Pattern 2: Parallel Search + Merge

Search images and documents in parallel, then merge results for synthesis.

from langchain_core.runnables import RunnableParallel, RunnableLambda


async def search_images_step(query: str) -> dict:
    result = await client.agent_search.images(query, limit=20)
    return {"image_ids": result.result_ids, "image_summary": result.summary}


async def search_docs_step(query: str) -> dict:
    result = await client.agent_search.documents(query, limit=10)
    return {"doc_ids": result.document_ids, "doc_summary": result.summary}


async def synthesize_step(data: dict) -> str:
    result = await client.agent_operations.synthesize(
        "Summarize all findings",
        image_ids=data["images"]["image_ids"],
        document_ids=data["docs"]["doc_ids"],
    )
    return result.summary


chain = (
    RunnableParallel(
        images=RunnableLambda(search_images_step),
        docs=RunnableLambda(search_docs_step),
    )
    | RunnableLambda(synthesize_step)
)


report = await chain.ainvoke("safety inspection Q4")

Pattern 3: Analysis Chain with Prompt Template

Search documents, then use a prompt template to extract structured analysis via the LLM.

from langchain_core.output_parsers import JsonOutputParser


async def search_docs(input: dict) -> dict:
    result = await client.agent_search.documents(input["query"], limit=10)
    return {
        "documents": result.summary,
        "doc_ids": result.document_ids,
        "analysis_type": input["analysis_type"],
    }


analysis_prompt = ChatPromptTemplate.from_template(
    "Analyze the following documents for {analysis_type}.\n\n"
    "Documents: {documents}\n\n"
    "Return a JSON object with: findings (list), risk_level (low/medium/high), "
    "and recommendations (list)."
)


chain = (
    RunnableLambda(search_docs)
    | analysis_prompt
    | model
    | JsonOutputParser()
)


analysis = await chain.ainvoke({
    "query": "compliance reports",
    "analysis_type": "regulatory compliance gaps",
})

Structured Output

Get typed Pydantic results from Aionvision + LLM using with_structured_output.

from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda


class InspectionReport(BaseModel):
    findings: list[str] = Field(description="Key findings from the inspection")
    severity: str = Field(description="Overall severity: low, medium, high, critical")
    recommendations: list[str] = Field(description="Recommended actions")
    compliant: bool = Field(description="Whether the inspection passes compliance")


async def gather_data(query: str) -> dict:
    images = await client.agent_search.images(query, limit=20)
    synthesis = await client.agent_operations.synthesize(
        "Summarize inspection findings", image_ids=images.result_ids
    )
    return {"data": synthesis.summary}


prompt = ChatPromptTemplate.from_template(
    "Based on the following inspection data, extract a structured report.\n\n"
    "{data}"
)


structured_model = ChatAnthropic(
    model="claude-sonnet-4-20250514"
).with_structured_output(InspectionReport)


chain = RunnableLambda(gather_data) | prompt | structured_model


# Returns an InspectionReport instance
report = await chain.ainvoke("site inspection photos")
print(report.severity)        # "medium"
print(report.recommendations) # ["Fix railing on floor 3", ...]

Streaming

LCEL chains and agents support streaming out of the box.

Chain Streaming

Stream token-by-token output from an LCEL chain.

# Using the search → summary chain from above
async for chunk in chain.astream("inspection photos from site A"):
    print(chunk, end="", flush=True)

Agent Event Streaming

Stream events from an agent, including tool calls and their results.

agent = create_react_agent(
    model=ChatAnthropic(model="claude-sonnet-4-20250514"),
    tools=[search_images, synthesize_report],
    prompt="You are a file intelligence assistant.",
)


async for event in agent.astream_events(
    {"messages": [{"role": "user", "content": "Find and summarize inspection data"}]},
    version="v2",
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)
    elif kind == "on_tool_start":
        print(f"\n[Calling {event['name']}...]")
    elif kind == "on_tool_end":
        print(f"[Done: {event['data']['output'][:100]}...]")

Complete Example — Document Analysis Workflow

A full end-to-end example: search documents and images in parallel, analyze them, extract structured findings, and generate a report. Uses LCEL chain composition with RunnableParallel and structured output.

import asyncio
from pydantic import BaseModel, Field
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel
from aion import AionVision


# ── Structured Output Schema ─────────────────────────
class ContractAnalysis(BaseModel):
    key_terms: list[str] = Field(description="Important terms and conditions")
    risks: list[str] = Field(description="Identified risks or concerns")
    expiration_dates: list[str] = Field(description="Contract expiration dates found")
    total_value: str | None = Field(description="Total contract value if mentioned")
    recommendation: str = Field(description="Overall recommendation")


# ── Aionvision Steps ─────────────────────────────────
async def search_contracts(input: dict) -> dict:
    result = await client.agent_search.documents(
        input["query"], limit=15, document_types=["pdf"]
    )
    return {
        "doc_ids": result.document_ids,
        "doc_summary": result.summary,
        "count": result.count,
    }


async def search_related_images(input: dict) -> dict:
    result = await client.agent_search.images(input["query"], limit=10)
    return {
        "image_ids": result.result_ids,
        "image_summary": result.summary,
    }


async def analyze_step(data: dict) -> dict:
    docs = data["documents"]
    images = data["images"]
    if docs["doc_ids"]:
        analysis = await client.agent_operations.analyze_documents(
            "Extract key terms, risks, and expiration dates",
            document_ids=docs["doc_ids"],
        )
    else:
        analysis = type("R", (), {"summary": "No documents found"})()
    return {
        "analysis": analysis.summary,
        "doc_count": docs["count"],
        "image_count": len(images["image_ids"]),
    }


# ── LLM Chain ────────────────────────────────────────
model = ChatAnthropic(model="claude-sonnet-4-20250514")
structured_model = model.with_structured_output(ContractAnalysis)


extraction_prompt = ChatPromptTemplate.from_template(
    "Based on the following document analysis, extract structured contract information.\n\n"
    "Documents analyzed: {doc_count}\n"
    "Related images: {image_count}\n\n"
    "Analysis:\n{analysis}"
)


# ── Full Chain ────────────────────────────────────────
chain = (
    RunnableParallel(
        documents=RunnableLambda(search_contracts),
        images=RunnableLambda(search_related_images),
    )
    | RunnableLambda(analyze_step)
    | extraction_prompt
    | structured_model
)


# ── Run ───────────────────────────────────────────────
async def main():
    global client
    async with AionVision.from_env() as client:
        result = await chain.ainvoke({
            "query": "vendor contracts expiring this year"
        })


        print(f"Key Terms: {result.key_terms}")
        print(f"Risks: {result.risks}")
        print(f"Expiration Dates: {result.expiration_dates}")
        print(f"Total Value: {result.total_value}")
        print(f"Recommendation: {result.recommendation}")


asyncio.run(main())

Best Practices

Use LCEL for Fixed Flows

When you know the steps upfront, LCEL chains are simpler and more predictable than agents. Reserve agents for tasks where the LLM should decide the path dynamically.

Write Clear Docstrings

The LLM uses tool docstrings to decide when to call each tool. Be specific about what the tool does, when to use it, and what it returns. Vague docstrings lead to incorrect tool selection.

Prefer create_react_agent

Unless you need fine-grained control over the tool call loop, create_react_agent handles it automatically. Use the manual loop only when you need to inspect or modify tool results between steps.

Use Structured Output

When results feed into other systems, use with_structured_output() for typed, parseable responses. This gives you Pydantic models instead of raw text, making downstream processing reliable.

What's Next?

LangGraph Integration

Stateful agents with custom state, human-in-the-loop, and full graph control

Pipelines

Server-side orchestration for known multi-step workflows

Standalone Agents

Direct access to individual search and operation agents

Client Configuration

SDK setup, authentication, and environment configuration