Examples Catalog¶

"""Basic LangChain agent with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy
from pathlib import Path

def langchain_agent():
    """Simple LangChain agent that uses OpenAI."""
    from langchain_openai import ChatOpenAI
    from langchain.prompts import ChatPromptTemplate

    # Create LLM
    llm = ChatOpenAI(model="gpt-4", temperature=0.7)

    # Create prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful AI assistant."),
        ("user", "{input}")
    ])

    # Create chain
    chain = prompt | llm

    # Run chain
    result = chain.invoke({"input": "Explain quantum computing in simple terms"})

    # Save output
    Path("/output/response.txt").write_text(result.content)

    return {"status": "success", "length": len(result.content)}

# Run agent in isolated container
result = run_agent(
    agent=langchain_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=2.0,
        memory_mb=1024,
        network=NetworkPolicy(
            disabled=False,
            allowed_endpoints=["api.openai.com:443"]
        ),
        pip_packages=["langchain", "langchain-openai"],
    )
)

print(f"Status: {result.return_value['status']}")
print(f"Output: {result.artifacts['response.txt']}")

"""LangChain RAG agent with vector store."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def rag_agent():
    """RAG agent with FAISS vector store."""
    from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    from langchain.vectorstores import FAISS
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain.chains import RetrievalQA
    from pathlib import Path

    # Load documents
    docs_path = Path("/workspace/documents")
    documents = []
    for file in docs_path.glob("*.txt"):
        documents.append(file.read_text())

    # Split documents
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    splits = text_splitter.create_documents(documents)

    # Create vector store
    embeddings = OpenAIEmbeddings()
    vectorstore = FAISS.from_documents(splits, embeddings)

    # Create QA chain
    llm = ChatOpenAI(model="gpt-4")
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        return_source_documents=True
    )

    # Query
    result = qa_chain({"query": "What are the main topics in these documents?"})

    # Save results
    Path("/output/answer.txt").write_text(result["result"])
    Path("/output/sources.txt").write_text(
        "\n\n".join([doc.page_content for doc in result["source_documents"]])
    )

    return {"status": "success", "sources": len(result["source_documents"])}

result = run_agent(
    agent=rag_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=4.0,
        memory_mb=2048,
        network=NetworkPolicy(disabled=False),
        pip_packages=["langchain", "langchain-openai", "faiss-cpu"],
    )
)

"""LangChain multi-agent system."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def researcher():
    """Research agent."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    llm = ChatOpenAI(model="gpt-4")
    research = llm.invoke("Research the latest developments in AI safety")

    Path("/output/research.txt").write_text(research.content)
    return {"status": "success"}

def writer():
    """Writer agent that uses research."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    research = Path("/workspace/research.txt").read_text()

    llm = ChatOpenAI(model="gpt-4")
    article = llm.invoke(f"Write an article based on: {research}")

    Path("/output/article.txt").write_text(article.content)
    return {"status": "success"}

# Run researcher
research_result = run_agent(
    agent=researcher,
    working_dir="./workspace",
    host_output_path="./output",
    policy=Policy(
        network=NetworkPolicy(disabled=False),
        pip_packages=["langchain", "langchain-openai"],
    )
)

# Copy research to workspace for writer
import shutil
shutil.copy("./output/research.txt", "./workspace/research.txt")

# Run writer
writer_result = run_agent(
    agent=writer,
    working_dir="./workspace",
    host_output_path="./output",
    policy=Policy(
        network=NetworkPolicy(disabled=False),
        pip_packages=["langchain", "langchain-openai"],
    )
)

print(f"Article: {writer_result.artifacts['article.txt']}")

"""Basic CrewAI crew with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def crewai_agent():
    """CrewAI crew for content creation."""
    from crewai import Agent, Task, Crew
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    llm = ChatOpenAI(model="gpt-4")

    # Define agents
    researcher = Agent(
        role="Researcher",
        goal="Research topics thoroughly",
        backstory="Expert researcher with attention to detail",
        llm=llm
    )

    writer = Agent(
        role="Writer",
        goal="Write engaging content",
        backstory="Professional content writer",
        llm=llm
    )

    # Define tasks
    research_task = Task(
        description="Research AI safety best practices",
        agent=researcher
    )

    writing_task = Task(
        description="Write an article about AI safety",
        agent=writer
    )

    # Create crew
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, writing_task],
        verbose=True
    )

    # Run crew
    result = crew.kickoff()

    # Save output
    Path("/output/article.txt").write_text(str(result))

    return {"status": "success", "length": len(str(result))}

result = run_agent(
    agent=crewai_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=4.0,
        memory_mb=2048,
        network=NetworkPolicy(disabled=False),
        pip_packages=["crewai", "langchain-openai"],
    )
)

"""AutoGPT agent with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def autogpt_agent():
    """AutoGPT agent for autonomous task completion."""
    from autogpt.agent import Agent
    from autogpt.config import Config
    from pathlib import Path

    # Configure AutoGPT
    config = Config()
    config.continuous_mode = False
    config.continuous_limit = 10

    # Create agent
    agent = Agent(
        ai_name="TaskAgent",
        memory=None,
        next_action_count=0,
        command_registry=None,
        config=config,
        system_prompt="You are an AI assistant that completes tasks autonomously.",
        triggering_prompt="Complete the given task efficiently."
    )

    # Run agent
    result = agent.run(["Research and summarize AI safety guidelines"])

    # Save output
    Path("/output/result.txt").write_text(str(result))

    return {"status": "success"}

result = run_agent(
    agent=autogpt_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=4.0,
        memory_mb=4096,
        network=NetworkPolicy(disabled=False),
        pip_packages=["autogpt"],
        timeout_seconds=600,
    )
)

"""LlamaIndex agent with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def llamaindex_agent():
    """LlamaIndex agent for document querying."""
    from llama_index import VectorStoreIndex, SimpleDirectoryReader
    from llama_index.llms import OpenAI
    from pathlib import Path

    # Load documents
    documents = SimpleDirectoryReader("/workspace/documents").load_data()

    # Create index
    llm = OpenAI(model="gpt-4")
    index = VectorStoreIndex.from_documents(documents, llm=llm)

    # Query
    query_engine = index.as_query_engine()
    response = query_engine.query("What are the main topics in these documents?")

    # Save output
    Path("/output/response.txt").write_text(str(response))

    return {"status": "success", "response_length": len(str(response))}

result = run_agent(
    agent=llamaindex_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=2.0,
        memory_mb=2048,
        network=NetworkPolicy(disabled=False),
        pip_packages=["llama-index", "openai"],
    )
)

"""Run Node.js agent with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

result = run_agent(
    agent=None,  # No Python callable
    working_dir="./workspace",
    policy=Policy(
        entrypoint=["node", "basic_agent.js"],
        network=NetworkPolicy(disabled=False),
        allowed_env_vars=["OPENAI_API_KEY"],
    )
)

print(f"Output: {result.artifacts['response.txt']}")

"""Run Go agent with Isolated Agents SDK."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

result = run_agent(
    agent=None,
    working_dir="./workspace",
    policy=Policy(
        entrypoint=["go", "run", "basic_agent.go"],
        network=NetworkPolicy(disabled=False),
        allowed_env_vars=["OPENAI_API_KEY"],
    )
)

"""Web scraping with analysis."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def scraping_agent():
    """Scrape website and analyze content."""
    import requests
    from bs4 import BeautifulSoup
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    # Scrape website
    response = requests.get("https://example.com")
    soup = BeautifulSoup(response.content, 'html.parser')
    text = soup.get_text()

    # Analyze with LLM
    llm = ChatOpenAI(model="gpt-4")
    analysis = llm.invoke(f"Analyze this content: {text[:2000]}")

    # Save results
    Path("/output/scraped.txt").write_text(text)
    Path("/output/analysis.txt").write_text(analysis.content)

    return {"status": "success", "content_length": len(text)}

result = run_agent(
    agent=scraping_agent,
    working_dir="./workspace",
    policy=Policy(
        network=NetworkPolicy(
            disabled=False,
            allowed_endpoints=["example.com:443", "api.openai.com:443"]
        ),
        pip_packages=["requests", "beautifulsoup4", "langchain-openai"],
    )
)

"""Data analysis with pandas and visualization."""

from isolated_agents_sdk import run_agent, Policy

def data_analysis_agent():
    """Analyze CSV data and create visualizations."""
    import pandas as pd
    import matplotlib.pyplot as plt
    from pathlib import Path

    # Load data
    df = pd.read_csv("/workspace/data.csv")

    # Analyze
    summary = df.describe()
    correlations = df.corr()

    # Create visualizations
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))

    df.hist(ax=axes[0, 0])
    axes[0, 0].set_title("Distributions")

    df.plot(kind='box', ax=axes[0, 1])
    axes[0, 1].set_title("Box Plots")

    correlations.plot(kind='bar', ax=axes[1, 0])
    axes[1, 0].set_title("Correlations")

    df.plot(kind='scatter', x=df.columns[0], y=df.columns[1], ax=axes[1, 1])
    axes[1, 1].set_title("Scatter Plot")

    plt.tight_layout()
    plt.savefig("/output/analysis.png", dpi=300)

    # Save summary
    Path("/output/summary.txt").write_text(summary.to_string())
    Path("/output/correlations.txt").write_text(correlations.to_string())

    return {"status": "success", "rows": len(df)}

result = run_agent(
    agent=data_analysis_agent,
    working_dir="./workspace",
    policy=Policy(
        cpu_cores=2.0,
        memory_mb=2048,
        pip_packages=["pandas", "matplotlib", "seaborn"],
    )
)

"""Code generation with validation."""

from isolated_agents_sdk import run_agent, Policy, NetworkPolicy

def code_generator():
    """Generate and validate Python code."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path
    import ast

    llm = ChatOpenAI(model="gpt-4")

    # Generate code
    prompt = """Generate a Python function that:
    1. Takes a list of numbers
    2. Filters out negative numbers
    3. Returns the sum of remaining numbers
    Include docstring and type hints."""

    code = llm.invoke(prompt).content

    # Validate syntax
    try:
        ast.parse(code)
        valid = True
    except SyntaxError:
        valid = False

    # Save code
    Path("/output/generated_code.py").write_text(code)
    Path("/output/validation.txt").write_text(f"Valid: {valid}")

    return {"status": "success", "valid": valid}

result = run_agent(
    agent=code_generator,
    working_dir="./workspace",
    policy=Policy(
        network=NetworkPolicy(disabled=False),
        pip_packages=["langchain-openai"],
    )
)

"""Example using all decorator types."""

from isolated_agents_sdk import (
    isolated_agent,
    policy,
    network,
    resources,
    dependencies,
    timeout,
    telemetry,
    retry,
    cache,
)

@isolated_agent(working_dir="./workspace")
@policy(memory_mb=2048, cpu_cores=2.0)
@network(enabled=True, allowed_endpoints=["api.openai.com:443"])
@resources(cpu_cores=2.0, memory_mb=2048)
@dependencies("langchain", "langchain-openai", "pandas")
@timeout(seconds=300)
@telemetry(enabled=True, level="INFO")
@retry(max_attempts=3, backoff=2.0)
@cache(ttl=3600)
def comprehensive_agent(query: str):
    """Agent with all decorators."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    llm = ChatOpenAI(model="gpt-4")
    result = llm.invoke(query)

    Path("/output/response.txt").write_text(result.content)
    return {"status": "success"}

# Use like a normal function
result = comprehensive_agent("Explain AI safety")
print(result)

"""Multi-agent pipeline example."""

from isolated_agents_sdk import isolated_agent, chain, parallel

@isolated_agent(working_dir="./workspace")
def researcher(topic: str):
    """Research agent."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    llm = ChatOpenAI(model="gpt-4")
    research = llm.invoke(f"Research: {topic}")
    Path("/output/research.txt").write_text(research.content)
    return {"status": "success"}

@isolated_agent(working_dir="./workspace")
def writer(research_file: str):
    """Writer agent."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    research = Path(research_file).read_text()
    llm = ChatOpenAI(model="gpt-4")
    article = llm.invoke(f"Write article: {research}")
    Path("/output/article.txt").write_text(article.content)
    return {"status": "success"}

@isolated_agent(working_dir="./workspace")
def editor(article_file: str):
    """Editor agent."""
    from langchain_openai import ChatOpenAI
    from pathlib import Path

    article = Path(article_file).read_text()
    llm = ChatOpenAI(model="gpt-4")
    edited = llm.invoke(f"Edit: {article}")
    Path("/output/final.txt").write_text(edited.content)
    return {"status": "success"}

# Chain agents together
@chain(agents=[researcher, writer, editor])
def content_pipeline(topic: str):
    """Complete content creation pipeline."""
    pass

# Run pipeline
result = content_pipeline("AI Safety")
print(f"Final: {result.artifacts['final.txt']}")