Integration Guide¶

import asyncio
from tinman import create_tinman, OperatingMode
from tinman.integrations import OpenAIClient

async def main():
    # Create model client
    client = OpenAIClient(api_key="sk-...")

    # Create Tinman instance
    tinman = await create_tinman(
        mode=OperatingMode.LAB,
        model_client=client,
        db_url="postgresql://localhost/tinman",
    )

    # Run research
    results = await tinman.research_cycle(
        focus="reasoning failures in code generation",
        max_hypotheses=5,
    )

    # Generate report
    report = await tinman.generate_report()
    print(report)

    await tinman.close()

asyncio.run(main())

from tinman.integrations import PipelineAdapter, FailureDetectionHook
from tinman.config.modes import OperatingMode

# Create adapter
adapter = PipelineAdapter(mode=OperatingMode.SHADOW)
adapter.register_hook(FailureDetectionHook())

# In your existing pipeline
async def my_llm_call(messages: list, model: str):
    # Pre-request hook
    ctx = adapter.create_context(messages=messages, model=model)
    ctx = await adapter.pre_request(ctx)

    # Your existing LLM call
    response = await your_model_client.complete(messages, model)

    # Post-request hook
    ctx.response = response
    ctx = await adapter.post_request(ctx)

    # Check for detected issues
    if "detected_failures" in ctx.metadata:
        log_potential_failures(ctx.metadata["detected_failures"])

    return response

from tinman import create_tinman
from tinman.core.event_bus import EventBus

# Create event bus
event_bus = EventBus()

# Subscribe to events
@event_bus.on("failure.discovered")
async def on_failure(event_data):
    await send_alert(
        channel="ai-reliability",
        message=f"New failure: {event_data['failure_class']}"
    )

@event_bus.on("intervention.proposed")
async def on_intervention(event_data):
    await create_ticket(
        title=f"Intervention proposed: {event_data['name']}",
        body=event_data['description']
    )

# Tinman will emit events as it works
tinman = await create_tinman(mode=OperatingMode.LAB)
tinman.event_bus = event_bus

from tinman.integrations.pipeline_adapter import HookPoint

class HookPoint(str, Enum):
    PRE_REQUEST = "pre_request"      # Before model call
    POST_REQUEST = "post_request"    # After model call
    PRE_TOOL = "pre_tool"           # Before tool execution
    POST_TOOL = "post_tool"         # After tool execution
    ON_ERROR = "on_error"           # On any error
    ON_COMPLETION = "on_completion"  # On task completion

from tinman.integrations.pipeline_adapter import (
    PipelineHook, HookPoint, HookResult, PipelineContext
)

class CustomValidationHook(PipelineHook):
    """Custom hook for validating model responses."""

    @property
    def name(self) -> str:
        return "custom_validation"

    @property
    def hook_points(self) -> list[HookPoint]:
        return [HookPoint.POST_REQUEST]

    async def execute(self,
                      hook_point: HookPoint,
                      context: PipelineContext) -> HookResult:
        # Validate response
        if context.response:
            content = str(context.response.get("content", ""))

            # Check for concerning patterns
            if "I cannot" in content and "error" in content.lower():
                # Log but don't block (in SHADOW mode)
                context.metadata["validation_warning"] = "Potential refusal pattern"
                return HookResult(
                    allow=True,
                    modified_context=context,
                    message="Logged potential issue"
                )

        return HookResult(allow=True)

from tinman.integrations.pipeline_adapter import LoggingHook

adapter = PipelineAdapter()
adapter.register_hook(LoggingHook())
# Logs all hook points

from tinman.integrations.pipeline_adapter import TokenLimitHook

adapter = PipelineAdapter()
adapter.register_hook(TokenLimitHook(max_tokens=50000))
# Blocks requests exceeding token limit

from tinman.integrations.pipeline_adapter import FailureDetectionHook

adapter = PipelineAdapter()
adapter.register_hook(FailureDetectionHook(
    failure_patterns=["error", "failed", "cannot", "unable"]
))
# Detects potential failures in responses

from tinman.integrations.pipeline_adapter import PipelineContext

@dataclass
class PipelineContext:
    id: str                           # Unique request ID
    mode: OperatingMode               # Current operating mode

    # Request data
    messages: list[dict]              # Input messages
    model: str                        # Target model
    tools: list[dict]                 # Available tools

    # Response data (after request)
    response: Optional[dict]          # Model response
    error: Optional[str]              # Error if any

    # Metadata (for passing data between hooks)
    metadata: dict[str, Any]          # Custom metadata
    started_at: datetime              # Request start time

# In LAB/PRODUCTION mode: raises PipelineBlocked exception
# In SHADOW mode: logs warning but continues

async def execute(self, hook_point, context) -> HookResult:
    if should_block(context):
        return HookResult(
            allow=False,
            message="Request blocked: exceeds safety threshold"
        )
    return HookResult(allow=True)

from tinman.integrations.model_client import ModelClient, ModelResponse

class ModelClient(ABC):
    """Abstract base class for model clients."""

    @property
    @abstractmethod
    def provider(self) -> str:
        """Provider name (e.g., 'openai', 'anthropic')."""
        pass

    @abstractmethod
    async def complete(self,
                       messages: list[dict[str, str]],
                       model: Optional[str] = None,
                       temperature: float = 0.7,
                       max_tokens: int = 4096,
                       tools: Optional[list[dict]] = None,
                       **kwargs) -> ModelResponse:
        """Send a completion request."""
        pass

    @abstractmethod
    async def stream(self,
                     messages: list[dict[str, str]],
                     model: Optional[str] = None,
                     **kwargs):
        """Stream a completion response."""
        pass

@dataclass
class ModelResponse:
    id: str                              # Response ID
    content: str                         # Generated content
    model: str                           # Model used

    # Token usage
    prompt_tokens: int                   # Input tokens
    completion_tokens: int               # Output tokens
    total_tokens: int                    # Total tokens

    # Tool calls
    tool_calls: list[dict[str, Any]]     # Tool call requests

    # Metadata
    finish_reason: str                   # Stop reason
    latency_ms: int                      # Response latency
    raw: Optional[dict[str, Any]]        # Raw response

from tinman.integrations import OpenAIClient

client = OpenAIClient(
    api_key="sk-...",                    # Or use OPENAI_API_KEY env var
    base_url="https://api.openai.com/v1",  # Optional custom endpoint
    organization="org-...",               # Optional organization
)

# Use with Tinman
tinman = await create_tinman(
    model_client=client,
    mode=OperatingMode.LAB,
)

from tinman.integrations import AnthropicClient

client = AnthropicClient(
    api_key="sk-ant-...",                # Or use ANTHROPIC_API_KEY env var
    base_url="https://api.anthropic.com",  # Optional custom endpoint
)

# Use with Tinman
tinman = await create_tinman(
    model_client=client,
    mode=OperatingMode.LAB,
)

from tinman.integrations import OpenRouterClient

client = OpenRouterClient(
    api_key="sk-or-...",  # Or set OPENROUTER_API_KEY env var
)

# Use DeepSeek (free tier available)
tinman = await create_tinman(model_client=client)

# Specify model explicitly
response = await client.complete(
    messages=[{"role": "user", "content": "Hello"}],
    model="deepseek-chat",  # Shorthand
    # or: model="deepseek/deepseek-chat"  # Full ID
)

from tinman.integrations import GroqClient

client = GroqClient(
    api_key="gsk_...",  # Or set GROQ_API_KEY env var
)

# Use Llama 3.1 70B (very fast!)
tinman = await create_tinman(model_client=client)

# Specify model
response = await client.complete(
    messages=[{"role": "user", "content": "Hello"}],
    model="llama-3.1-70b",
)

from tinman.integrations import OllamaClient

# No API key needed!
client = OllamaClient(
    base_url="http://localhost:11434/v1",  # Default
)

# Use local Llama
tinman = await create_tinman(model_client=client)

# Use Qwen
response = await client.complete(
    messages=[{"role": "user", "content": "Hello"}],
    model="qwen2.5",
)

# List locally available models
models = await client.list_local_models()
print(models)  # ['llama3.1', 'qwen2.5', ...]

from tinman.integrations import TogetherClient

client = TogetherClient(
    api_key="...",  # Or set TOGETHER_API_KEY env var
)

# Use DeepSeek V3
tinman = await create_tinman(model_client=client)

response = await client.complete(
    messages=[{"role": "user", "content": "Hello"}],
    model="deepseek-v3",
)

from tinman.integrations.model_client import ModelClient, ModelResponse

class CustomClient(ModelClient):
    """Client for custom LLM endpoint."""

    def __init__(self, endpoint: str, api_key: str):
        super().__init__(api_key=api_key)
        self.endpoint = endpoint

    @property
    def provider(self) -> str:
        return "custom"

    async def complete(self,
                       messages: list[dict[str, str]],
                       model: Optional[str] = None,
                       temperature: float = 0.7,
                       max_tokens: int = 4096,
                       tools: Optional[list[dict]] = None,
                       **kwargs) -> ModelResponse:
        # Your implementation
        async with aiohttp.ClientSession() as session:
            response = await session.post(
                self.endpoint,
                json={
                    "messages": messages,
                    "temperature": temperature,
                    "max_tokens": max_tokens,
                },
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            data = await response.json()

        return ModelResponse(
            content=data["content"],
            model=data.get("model", "custom"),
            prompt_tokens=data.get("usage", {}).get("prompt_tokens", 0),
            completion_tokens=data.get("usage", {}).get("completion_tokens", 0),
            total_tokens=data.get("usage", {}).get("total_tokens", 0),
            tool_calls=data.get("tool_calls", []),
            finish_reason=data.get("finish_reason", ""),
            latency_ms=0,  # Calculate from timing
            raw=data,
        )

    async def stream(self, messages, **kwargs):
        # Your streaming implementation
        pass

from tinman import Tinman, create_tinman, OperatingMode
from tinman.integrations import OpenAIClient
from tinman.core.approval_handler import ApprovalMode

# Method 1: Factory function (recommended)
tinman = await create_tinman(
    mode=OperatingMode.LAB,
    model_client=OpenAIClient(),
    db_url="postgresql://localhost/tinman",
)

# Method 2: Manual initialization
tinman = Tinman(
    model_client=OpenAIClient(),
    mode=OperatingMode.LAB,
    approval_mode=ApprovalMode.AUTO_APPROVE,  # For automated systems
    auto_approve_in_lab=True,
)
await tinman.initialize(db_url="postgresql://localhost/tinman")

from tinman.agents import (
    HypothesisEngine,
    ExperimentArchitect,
    ExperimentExecutor,
    FailureDiscoveryAgent,
)
from tinman.memory import MemoryGraph
from tinman.reasoning import LLMBackbone

# Just hypothesis generation
hypothesis_engine = HypothesisEngine(
    graph=your_graph,
    llm_backbone=your_llm,
)
hypotheses = await hypothesis_engine.run(context)

# Just failure classification
failure_discovery = FailureDiscoveryAgent(
    graph=your_graph,
    llm_backbone=your_llm,
)
failures = await failure_discovery.run(context, results=experiment_results)

from tinman import Tinman
from tinman.core.approval_handler import ApprovalMode

# Create with interactive approvals
tinman = Tinman(
    model_client=client,
    approval_mode=ApprovalMode.INTERACTIVE,
    auto_approve_in_lab=False,  # Require all approvals
)

# Register custom UI callback
async def my_approval_ui(context):
    """Custom approval UI."""
    # Show in your UI
    approved = await show_approval_dialog(
        title=context.action_name,
        description=context.description,
        risk_tier=context.risk_tier,
    )
    return approved

tinman.register_approval_ui(my_approval_ui)

from tinman.core.event_bus import EventBus

event_bus = EventBus()

# Subscribe to events
@event_bus.on("hypothesis.generated")
async def handle_hypothesis(data):
    print(f"New hypothesis: {data['target_surface']}")

# Or subscribe programmatically
def my_handler(data):
    log_event("hypothesis.generated", data)

event_bus.subscribe("hypothesis.generated", my_handler)

# Unsubscribe
event_bus.unsubscribe("hypothesis.generated", my_handler)

from tinman import create_tinman
from tinman.core.event_bus import EventBus

# Create shared event bus
event_bus = EventBus()

# Subscribe handlers
@event_bus.on("failure.discovered")
async def alert_on_failure(data):
    if data.get("severity") in ["S3", "S4"]:
        await pagerduty.create_incident(
            title=f"Critical AI Failure: {data['failure_class']}",
            severity="high",
            details=data,
        )

@event_bus.on("experiment.completed")
async def log_experiment(data):
    await metrics.record("tinman.experiments", 1, tags={
        "failures": data["failures_triggered"],
    })

# Use with Tinman
tinman = await create_tinman(mode=OperatingMode.LAB)
tinman.event_bus = event_bus

from tinman import create_tinman

tinman = await create_tinman(
    db_url="postgresql://user:pass@localhost:5432/tinman",
)

# Local development
db_url = "postgresql://localhost/tinman"

# With credentials
db_url = "postgresql://tinman_user:secret@localhost:5432/tinman"

# Remote server with SSL
db_url = "postgresql://user:pass@prod-db.example.com:5432/tinman?sslmode=require"

# SQLite for testing
db_url = "sqlite:///tinman.db"
db_url = "sqlite:///:memory:"  # In-memory

from tinman.db.connection import DatabaseConnection

# Create connection
db = DatabaseConnection("postgresql://localhost/tinman")
db.create_tables()

# Use session context manager
with db.session() as session:
    # Query operations
    results = session.execute(query).fetchall()

    # The session commits on exit

from tinman.db.connection import DatabaseConnection

db = DatabaseConnection(db_url)

# Create all tables
db.create_tables()

# The memory graph schema includes:
# - nodes table (with temporal validity)
# - edges table (with relationships)
# - indexes for common queries

from fastapi import FastAPI, BackgroundTasks
from contextlib import asynccontextmanager
from tinman import create_tinman, OperatingMode
from tinman.integrations import OpenAIClient

# Global Tinman instance
tinman = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global tinman
    tinman = await create_tinman(
        mode=OperatingMode.SHADOW,
        model_client=OpenAIClient(),
        db_url="postgresql://localhost/tinman",
    )
    yield
    await tinman.close()

app = FastAPI(lifespan=lifespan)

@app.post("/api/llm/complete")
async def complete(request: CompletionRequest, background_tasks: BackgroundTasks):
    # Your normal completion logic
    response = await your_llm_call(request)

    # Background analysis with Tinman
    background_tasks.add_task(analyze_with_tinman, request, response)

    return response

async def analyze_with_tinman(request, response):
    """Run Tinman analysis in background."""
    from tinman.integrations import record_llm_interaction
    record_llm_interaction(
        adapter=tinman.pipeline_adapter,
        messages=request.messages,
        model=request.model,
        response_text=response.content,
    )

@app.get("/api/tinman/status")
async def tinman_status():
    return tinman.get_state()

@app.get("/api/tinman/report")
async def tinman_report():
    return await tinman.generate_report(format="json")

@app.post("/api/tinman/research")
async def start_research(focus: str = None):
    results = await tinman.research_cycle(focus=focus)
    return {"status": "complete", "findings": len(results["failures"])}

from tinman.integrations import TinmanLangChainCallbackHandler

# Usage
from langchain.llms import OpenAI

llm = OpenAI(callbacks=[TinmanLangChainCallbackHandler()])
response = llm.invoke("Your prompt here")

from tinman.integrations import TinmanCrewHook

crew_hook = TinmanCrewHook()

# Wire into CrewAI task callbacks / events:
# crew_hook.on_task_start(task)
# crew_hook.on_task_end(task, output)
# crew_hook.on_task_error(task, error)

# settings.py
TINMAN_CONFIG = {
    "mode": "shadow",
    "db_url": "postgresql://localhost/tinman",
}

# services/tinman_service.py
import asyncio
from django.conf import settings
from tinman import create_tinman, OperatingMode
from tinman.integrations import OpenAIClient

class TinmanService:
    _instance = None
    _tinman = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    async def get_tinman(self):
        if self._tinman is None:
            config = settings.TINMAN_CONFIG
            self._tinman = await create_tinman(
                mode=OperatingMode(config["mode"]),
                model_client=OpenAIClient(),
                db_url=config["db_url"],
            )
        return self._tinman

    def run_research_sync(self, focus=None):
        """Synchronous wrapper for Django views."""
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            tinman = loop.run_until_complete(self.get_tinman())
            return loop.run_until_complete(
                tinman.research_cycle(focus=focus)
            )
        finally:
            loop.close()

# views.py
from django.http import JsonResponse
from .services.tinman_service import TinmanService

def tinman_status(request):
    service = TinmanService()
    loop = asyncio.new_event_loop()
    tinman = loop.run_until_complete(service.get_tinman())
    state = tinman.get_state()
    loop.close()
    return JsonResponse(state)

"""
Complete example: FastAPI service with Tinman integration.
"""
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional

from tinman import create_tinman, OperatingMode
from tinman.integrations import OpenAIClient, PipelineAdapter, FailureDetectionHook
from tinman.core.event_bus import EventBus

# Models
class CompletionRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4"
    temperature: float = 0.7

class ResearchRequest(BaseModel):
    focus: Optional[str] = None
    max_hypotheses: int = 5

# Global state
tinman = None
event_bus = EventBus()
adapter = None

# Event handlers
@event_bus.on("failure.discovered")
async def on_failure(data):
    print(f"ALERT: New failure discovered - {data['failure_class']}")
    # In production: send to monitoring system

@asynccontextmanager
async def lifespan(app: FastAPI):
    global tinman, adapter

    # Initialize Tinman
    client = OpenAIClient()
    tinman = await create_tinman(
        mode=OperatingMode.SHADOW,
        model_client=client,
        db_url="postgresql://localhost/tinman",
    )
    tinman.event_bus = event_bus

    # Initialize pipeline adapter
    adapter = PipelineAdapter(mode=OperatingMode.SHADOW)
    adapter.register_hook(FailureDetectionHook())

    yield

    await tinman.close()

app = FastAPI(
    title="AI Service with Tinman",
    lifespan=lifespan,
)

@app.post("/v1/completions")
async def create_completion(
    request: CompletionRequest,
    background_tasks: BackgroundTasks,
):
    """LLM completion endpoint with Tinman monitoring."""
    # Create pipeline context
    ctx = adapter.create_context(
        messages=request.messages,
        model=request.model,
    )

    # Pre-request hook
    ctx = await adapter.pre_request(ctx)

    # Your actual LLM call
    try:
        response = await tinman.model_client.complete(
            messages=request.messages,
            model=request.model,
            temperature=request.temperature,
        )
    except Exception as e:
        ctx.error = str(e)
        await adapter.on_error(ctx)
        raise HTTPException(status_code=500, detail=str(e))

    # Post-request hook
    ctx.response = {"content": response.content}
    ctx = await adapter.post_request(ctx)

    # Log any detected issues
    if "detected_failures" in ctx.metadata:
        background_tasks.add_task(
            log_potential_issues,
            ctx.metadata["detected_failures"],
            ctx.id,
        )

    return {
        "id": response.id,
        "content": response.content,
        "model": response.model,
        "usage": {
            "prompt_tokens": response.prompt_tokens,
            "completion_tokens": response.completion_tokens,
            "total_tokens": response.total_tokens,
        }
    }

async def log_potential_issues(issues: list, request_id: str):
    """Background task to log potential issues."""
    for issue in issues:
        print(f"[{request_id}] Potential issue detected: {issue}")

@app.post("/v1/research")
async def start_research(request: ResearchRequest):
    """Trigger a research cycle."""
    results = await tinman.research_cycle(
        focus=request.focus,
        max_hypotheses=request.max_hypotheses,
    )

    return {
        "hypotheses": len(results["hypotheses"]),
        "experiments": len(results["experiments"]),
        "failures": len(results["failures"]),
        "interventions": len(results["interventions"]),
    }

@app.get("/v1/tinman/status")
async def get_status():
    """Get Tinman status."""
    return tinman.get_state()

@app.get("/v1/tinman/report")
async def get_report(format: str = "markdown"):
    """Get research report."""
    report = await tinman.generate_report(format=format)
    return {"report": report}

@app.get("/v1/tinman/approvals/pending")
async def get_pending_approvals():
    """Get pending approval requests."""
    return tinman.get_pending_approvals()

@app.get("/health")
async def health():
    """Health check endpoint."""
    return await tinman.health_check()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)