LLM Provider Integration¶

from lionagi import Branch, iModel

# Explicit provider + model
model = iModel(provider="openai", model="gpt-4.1-mini")

# Slash-syntax auto-detects provider from the model string
model = iModel(model="openai/gpt-4.1-mini")

# Branch uses the model for all LLM operations
branch = Branch(chat_model=model)

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="openai", model="gpt-4.1-mini")
    )
    result = await branch.chat(
        instruction="Explain the difference between async and sync programming."
    )
    print(result)

asyncio.run(main())

# Responses endpoint
model = iModel(provider="openai", model="gpt-4.1-mini", endpoint="response")

import asyncio
from pydantic import BaseModel, Field
from lionagi import Branch, iModel

class SentimentResult(BaseModel):
    sentiment: str = Field(description="positive, negative, or neutral")
    confidence: float = Field(description="Confidence score 0-1")

async def main():
    branch = Branch(
        chat_model=iModel(provider="openai", model="gpt-4.1-mini")
    )
    result = await branch.operate(
        instruction="Analyze the sentiment of this review.",
        context="The product exceeded all my expectations!",
        operative_model=SentimentResult,
    )
    print(result)  # Operative with .output containing a SentimentResult

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="anthropic", model="claude-sonnet-4-20250514")
    )
    result = await branch.chat(
        instruction="Analyze this code for potential issues.",
        context="def divide(a, b): return a / b",
    )
    print(result)

asyncio.run(main())

result = await branch.chat(
    instruction="Summarize this long document.",
    context=long_document,
    cache_control=True,
)

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="gemini", model="gemini-2.5-flash")
    )
    result = await branch.chat(
        instruction="Compare Python and Rust for systems programming."
    )
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="ollama", model="llama3.2:3b")
    )
    result = await branch.chat(
        instruction="Explain how transformers work in machine learning."
    )
    print(result)

asyncio.run(main())

model = iModel(
    provider="ollama",
    model="llama3.2:3b",
    base_url="http://my-server:11434/v1",
)

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="perplexity", model="sonar")
    )
    result = await branch.chat(
        instruction="What are the latest developments in quantum computing?"
    )
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(provider="groq", model="llama-3.3-70b-versatile")
    )
    result = await branch.chat(
        instruction="Fast inference test -- explain recursion in one paragraph."
    )
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(
            provider="openrouter",
            model="google/gemini-2.5-flash",
        )
    )
    result = await branch.chat(instruction="Hello from OpenRouter!")
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    branch = Branch(
        chat_model=iModel(
            provider="nvidia_nim",
            model="meta/llama3-8b-instruct",
        )
    )
    result = await branch.chat(instruction="Explain GPU parallelism.")
    print(result)

asyncio.run(main())

embed_model = iModel(
    provider="nvidia_nim",
    endpoint="embed",
    model="nvidia/nv-embed-v1",
)

from lionagi import iModel

search_model = iModel(provider="exa", endpoint="search")

from lionagi import iModel

model = iModel(provider="claude_code", model="sonnet")

# Check if a model uses a CLI endpoint
model.is_cli  # True

import asyncio
from lionagi import Branch, iModel

async def main():
    model = iModel(
        provider="claude_code",
        model="sonnet",                          # "sonnet" or "opus"
        system_prompt="You are a code reviewer.", # system instructions
        permission_mode="bypassPermissions",      # skip approval prompts
        allowed_tools=["Read", "Grep", "Glob", "Bash"],
        max_turns=10,                            # conversation turn limit
    )
    branch = Branch(chat_model=model, name="reviewer")
    result = await branch.communicate("Review src/auth.py for security issues")
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    model = iModel(
        provider="gemini_code",
        model="gemini-2.5-pro",
        sandbox=True,                      # safety sandboxing (default)
        approval_mode="auto_edit",         # "suggest", "auto_edit", "full_auto"
    )
    branch = Branch(chat_model=model, name="gemini_agent")
    result = await branch.communicate("Analyze the project structure")
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    model = iModel(
        provider="codex",
        model="gpt-5.3-codex",
        sandbox="workspace-write",         # "read-only", "workspace-write", "danger-full-access"
        full_auto=True,                    # auto-approve with workspace-write sandbox
    )
    branch = Branch(chat_model=model, name="codex_agent")
    result = await branch.communicate("Fix the failing tests in this project")
    print(result)

asyncio.run(main())

model = iModel(provider="claude_code", model="sonnet")

# Normal: session resume is automatic
await branch.communicate("First task")
await branch.communicate("Follow-up")  # resumes same session

# Session too long -- rotate to fresh instance, carry context
branch.chat_model = iModel(provider="claude_code", model="sonnet")
recent = list(branch.msgs.progression)[-30:]
await branch.communicate("Continue from here.", progression=recent)

# Fresh copy, independent session
new_model = model.copy()

# Copy that resumes the same CLI session
resumed = model.copy(share_session=True)

model = iModel(provider="claude_code", model="sonnet")

model.endpoint.update_handlers(
    on_text=lambda chunk: print(f"Text: {chunk.text}"),
    on_tool_use=lambda chunk: print(f"Tool: {chunk}"),
    on_thinking=lambda chunk: print(f"Thinking: {chunk.text}"),
    on_final=lambda session: print(f"Done: {session.result}"),
)

from lionagi import iModel

# These two are equivalent:
m1 = iModel(provider="openai", model="gpt-4.1-mini")
m2 = iModel(model="openai/gpt-4.1-mini")

# Works with any provider
m3 = iModel(model="anthropic/claude-sonnet-4-20250514")
m4 = iModel(model="groq/llama-3.3-70b-versatile")
m5 = iModel(model="gemini/gemini-2.5-flash")

import asyncio
from lionagi import Branch, iModel

async def main():
    model = iModel(
        provider="together",
        model="meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
        base_url="https://api.together.xyz/v1",
        api_key="your-together-key",
    )
    branch = Branch(chat_model=model)
    result = await branch.chat(instruction="Hello from a custom provider!")
    print(result)

asyncio.run(main())

import asyncio
from lionagi import Branch, iModel

async def main():
    # iModel context manager -- stops the rate-limited executor on exit
    async with iModel(provider="openai", model="gpt-4.1-mini") as model:
        branch = Branch(chat_model=model)
        result = await branch.chat(instruction="Hello!")
        print(result)

    # Branch context manager -- flushes logs on exit
    async with Branch(
        chat_model=iModel(provider="openai", model="gpt-4.1-mini")
    ) as branch:
        result = await branch.chat(instruction="Hello again!")
        print(result)

asyncio.run(main())

from lionagi import iModel

model = iModel(provider="openai", model="gpt-4.1-mini")

# Fresh copy, no shared state
model2 = model.copy()

# For CLI providers: share_session=True carries over the session ID
cli_model = iModel(provider="claude_code", model="sonnet")
cli_model2 = cli_model.copy(share_session=True)

from lionagi import iModel

model = iModel(
    provider="openai",
    model="gpt-4.1-mini",
    queue_capacity=100,            # Max queued requests (default: 100, CLI: 10)
    capacity_refresh_time=60,      # Seconds between capacity refreshes
    limit_requests=50,             # Max requests per cycle
    limit_tokens=100_000,          # Max tokens per cycle
    concurrency_limit=10,          # Max concurrent streaming requests
)

from lionagi import Branch, iModel

branch = Branch(
    chat_model=iModel(provider="openai", model="gpt-4.1-mini"),
    parse_model=iModel(provider="anthropic", model="claude-sonnet-4-20250514"),
)

fast_model = iModel(provider="groq", model="llama-3.3-70b-versatile")

# Use the fast model just for this one call
result = await branch.chat(
    instruction="Quick question.",
    imodel=fast_model,
)