portainer-core/plans/active/unified-agent-architecture.md

Unified Agent Architecture Plan

Date: 2025-11-23 Objective: Build a single intelligent agent that handles all tool routing, multi-modal processing, and agentic reasoning internally, exposing one simple chat endpoint to any UI

Vision

Instead of configuring functions in Open WebUI (or any other UI), the Core API becomes an intelligent orchestrator that:

1. Accepts simple chat messages - Just like talking to ChatGPT
2. Internally routes to specialized tools/models - Infrastructure management, web search, code execution, etc.
3. Streams reasoning/thinking - Shows what it's doing ("Searching the web...", "Querying database...", "Using expert model...")
4. Returns unified responses - Combines results from multiple sources transparently

Benefits

✅ UI-agnostic - Works with Open WebUI, CLI, mobile apps, any client ✅ No configuration needed - Users just chat naturally ✅ Transparent reasoning - See what's happening under the hood ✅ Tool discovery - Agent decides when to use tools, not manual triggers ✅ Multi-modal support - Handle text, images, code, infrastructure queries ✅ Expert model routing - Use small models for simple tasks, large for complex

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                     User Interface                          │
│         (Open WebUI, CLI, Mobile App, etc.)                 │
└──────────────────────┬──────────────────────────────────────┘
                       │ Simple chat: "Deploy nginx proxy"
                       ↓
┌─────────────────────────────────────────────────────────────┐
│              Core API - Unified Agent                       │
│  /v1/chat/completions (OpenAI-compatible endpoint)          │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ↓
┌─────────────────────────────────────────────────────────────┐
│           Agent Orchestrator (PydanticAI)                   │
│  ┌──────────────────────────────────────────────┐           │
│  │  Reasoning Loop:                             │           │
│  │  1. Analyze user intent                      │           │
│  │  2. Select appropriate tool(s)               │           │
│  │  3. Execute tool calls                       │           │
│  │  4. Synthesize results                       │           │
│  │  5. Stream thinking/reasoning                │           │
│  └──────────────────────────────────────────────┘           │
└──────────────────────┬──────────────────────────────────────┘
                       │
        ┌──────────────┼──────────────┬──────────────┐
        │              │              │              │
        ↓              ↓              ↓              ↓
┌──────────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐
│ Tool Catalog │ │ Models   │ │ Memory   │ │ Knowledge    │
│              │ │          │ │          │ │              │
│ • Infra Mgmt │ │ • Gemma  │ │ • Qdrant │ │ • Web Search │
│ • Web Scrape │ │ • Codestral│ │ • Buffer│ │ • Docs       │
│ • File Ops   │ │ • Mistral│ │          │ │              │
│ • Code Exec  │ │          │ │          │ │              │
└──────────────┘ └──────────┘ └──────────┘ └──────────────┘

Implementation Options

Option 1: PydanticAI (Current Implementation)

Pros:

• Type-safe tool definitions with Pydantic models
• Built-in streaming support with structured output
• Native Ollama integration via HTTP API
• Lightweight and minimal dependencies
• Clear separation of concerns with dependency injection
• Excellent debugging with structured validation

Cons:

• Relatively new framework (less established patterns)
• Manual agent loop implementation required
• Less built-in state management compared to stateful frameworks

Example flow:

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

class DeployServiceParams(BaseModel):
    service_name: str
    compose_yaml: str

class WebSearchParams(BaseModel):
    query: str

agent = Agent(
    model="ollama:mistral-tools:7b",
    system_prompt="You are a homelab infrastructure assistant...",
    result_type=str
)

@agent.tool
async def deploy_service(ctx: RunContext[None], params: DeployServiceParams) -> str:
    """Deploy a containerized service via Portainer"""
    return await portainer_client.deploy_stack(
        params.service_name,
        params.compose_yaml
    )

@agent.tool
async def web_search(ctx: RunContext[None], params: WebSearchParams) -> str:
    """Search the web and extract content"""
    return await scraper.scrape(params.query)

# Streaming with reasoning
async with agent.run_stream(user_message) as stream:
    async for chunk in stream.stream_text():
        if chunk.type == "tool_call":
            yield f"data: {json.dumps({'tool': chunk.tool_name})}\n\n"
        elif chunk.type == "text":
            yield f"data: {json.dumps({'content': chunk.content})}\n\n"

Option 2: Custom Agent Loop

Pros:

• Full control over behavior
• Minimal dependencies
• Optimized for specific use case
• Easier to debug

Cons:

• More code to maintain
• Need to implement tool calling protocol
• Reinventing some wheels
• Manual type validation

Example flow:

class UnifiedAgent:
    def __init__(self):
        self.tools = ToolCatalog()
        self.model = OllamaClient()

    async def process(self, user_message: str):
        # 1. Intent analysis
        yield {"type": "thinking", "content": "Analyzing your request..."}
        intent = await self.analyze_intent(user_message)

        # 2. Tool selection
        if intent.requires_tool:
            yield {"type": "thinking", "content": f"Using {intent.tool_name}..."}
            tool_result = await self.tools.execute(intent.tool_name, intent.params)

        # 3. Response generation
        yield {"type": "thinking", "content": "Generating response..."}
        response = await self.model.generate(context=tool_result)

        yield {"type": "content", "content": response}

Option 3: Hybrid (PydanticAI + Custom Extensions)

Use PydanticAI's agent framework with custom enhancements:

• Leverage type-safe tool definitions
• Add custom routing logic for multi-model selection
• Enhanced streaming control for reasoning output
• Custom dependency injection for context management

Recommended Approach: PydanticAI with Custom Extensions

Phase 1: Core Agent (Week 1)

• Set up PydanticAI agent with basic tools
• Implement streaming with reasoning output
• Wire up existing infrastructure tools
• Test with simple queries

Phase 2: Advanced Routing (Week 2)

• Multi-model routing (small for simple, large for complex)
• Parallel tool execution via async tools
• Error handling and retries with custom logic
• Context management using RunContext

Phase 3: Multi-Modal (Week 3)

• Image analysis (if needed)
• Code execution sandbox
• File operations
• Database queries

Tool Catalog Design

Tier 1: Infrastructure Tools (Existing)

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

class ListServicesResult(BaseModel):
    services: List[Dict]

class DeployServiceParams(BaseModel):
    name: str
    compose: str

@agent.tool
async def list_services(ctx: RunContext[None]) -> ListServicesResult:
    """List all running Docker services"""
    services = await portainer_client.list_containers()
    return ListServicesResult(services=services)

@agent.tool
async def deploy_service(ctx: RunContext[None], params: DeployServiceParams) -> str:
    """Deploy a new service from Docker Compose YAML"""
    return await portainer_client.deploy_stack(params.name, params.compose)

@agent.tool
async def create_proxy(ctx: RunContext[None], domain: str, target: str) -> str:
    """Create Nginx reverse proxy for a service"""
    return await npm_client.create_proxy_host(domain, target)

@agent.tool
async def check_service_health(ctx: RunContext[None], service: str) -> Dict:
    """Check if a service is healthy"""
    return await kuma_client.get_monitor_status(service)

Tier 2: Knowledge Tools

@agent.tool
async def web_search(ctx: RunContext[None], query: str) -> str:
    """Search the web and extract main content"""
    return await scraper.scrape_url(query)

@agent.tool
async def query_memory(ctx: RunContext[None], question: str) -> List[str]:
    """Search conversation history for relevant context"""
    return await memory.semantic_search(question)

@agent.tool
async def read_documentation(ctx: RunContext[None], topic: str) -> str:
    """Read project documentation"""
    docs_path = f"/docs/{topic}.md"
    return read_file(docs_path)

Tier 3: Execution Tools (Future)

@agent.tool
async def execute_python(ctx: RunContext[None], code: str) -> str:
    """Execute Python code in sandbox"""
    # Future: Integrate code interpreter
    pass

@agent.tool
async def query_database(ctx: RunContext[None], sql: str) -> List[Dict]:
    """Query PostgreSQL database"""
    # Future: Safe SQL execution
    pass

Streaming Reasoning Output

SSE Format for Transparency

# Stream format
{
    "type": "thinking",  # or "tool_call", "content", "error"
    "content": "Searching the web for nginx configuration...",
    "tool": "web_search",  # optional, if type is tool_call
    "model": "mistral-tools:7b"    # optional, which model is being used
}

# Example stream
data: {"type": "thinking", "content": "Analyzing your request..."}

data: {"type": "thinking", "content": "Detected infrastructure task"}

data: {"type": "tool_call", "tool": "list_services", "content": "Checking current services..."}

data: {"type": "thinking", "content": "Found 22 running services"}

data: {"type": "thinking", "content": "Using expert model for response..."}

data: {"type": "model_switch", "from": "gemma:2b", "to": "mistral:7b"}

data: {"type": "content", "content": "Here are your running services:\n\n..."}

data: [DONE]

Open WebUI Integration

Open WebUI already supports streaming, we just need to format it correctly:

// Open WebUI will render thinking/reasoning in a collapsible section
// Standard content renders as usual

Model Routing Strategy

Intent-Based Routing

from pydantic_ai import Agent

class ModelRouter:
    MODELS = {
        "simple": "ollama:gemma:2b",      # Fast, <100 tokens
        "general": "ollama:gemma:7b",     # Balanced
        "expert": "ollama:mistral:7b",    # Complex reasoning
        "code": "ollama:codestral:latest" # Code tasks
    }

    async def select_model(self, message: str, context: str) -> str:
        # Use lightweight model for routing decision
        routing_agent = Agent(
            model="ollama:gemma:2b",
            result_type=str,
            system_prompt="""Analyze this request and categorize:

            Categories:
            - simple: Greetings, basic facts, short answers
            - general: Normal conversation, explanations
            - expert: Complex reasoning, multi-step problems
            - code: Programming tasks, debugging

            Return ONLY the category.
            """
        )

        result = await routing_agent.run(f"User: {message}\nContext: {context}")
        return self.MODELS[result.data.strip()]

Next Steps

1. Enhance PydanticAI agent (2-3 hours)

   • Add more infrastructure tools
   • Improve streaming with reasoning output
   • Test with complex queries

2. Integrate remaining tools (3-4 hours)

   • Migrate all infrastructure controller tools
   • Add web scraper tool improvements
   • Test multi-tool workflows

3. Model routing enhancements (2 hours)

   • Refine intent analysis
   • Add model selection metrics
   • Test performance improvements

4. Production hardening (2 hours)

   • Enhanced error handling
   • Rate limiting
   • Logging and monitoring
   • Update API documentation

Total effort: ~12-15 hours (1-2 weeks of focused work)

Success Criteria

✅ User can chat naturally without configuring functions ✅ Agent automatically uses tools when appropriate ✅ Streaming shows what the agent is doing ✅ Works with Open WebUI without changes ✅ Can be used from CLI/API directly ✅ Performance is acceptable (<5s for tool-using responses) ✅ Errors are handled gracefully

Example User Flows

Flow 1: Infrastructure Query

User: "What services are currently running?"

[Thinking: Analyzing request...]
[Thinking: Detected infrastructure query]
[Tool Call: list_services - Fetching service list...]
[Thinking: Processing results...]
[Content: You have 22 services running:
- ollama (healthy)
- core-api (healthy)
- ...]

Flow 2: Complex Task

User: "Deploy an nginx proxy for my new blog at blog.schweitz.net"

[Thinking: Breaking down the task...]
[Thinking: Need to deploy nginx and configure NPM]
[Tool Call: deploy_service - Deploying nginx container...]
[Tool Call: create_proxy - Creating proxy host...]
[Thinking: Configuring SSL certificate...]
[Content: Done! Your blog is now accessible at https://blog.schweitz.net
- Nginx container: running
- SSL certificate: active
- Health check: passing]

Flow 3: Knowledge Query

User: "How do I configure Headscale?"

[Thinking: Checking documentation...]
[Tool Call: read_documentation(headscale)]
[Thinking: Extracting relevant steps...]
[Content: To configure Headscale on tower-of-joy:

1. Create a user: `headscale users create homelab`
2. Generate auth key: `headscale preauthkeys create...`
...]

Technology Stack

• Agent Framework: PydanticAI
• LLM Integration: Native Ollama SDK (HTTP API)
• Tool Framework: PydanticAI Tools with Pydantic validation
• Streaming: SSE (Server-Sent Events)
• State Management: RunContext dependency injection
• Memory: Existing Qdrant integration

Risk Mitigation

Risk: PydanticAI is relatively new

• Mitigation: Strong typing provides safety, active development community

Risk: Tool calling may be slow

• Mitigation: Async tools enable parallel execution, caching, optimized tools

Risk: Reasoning output may be verbose

• Mitigation: Configurable verbosity, collapsible UI elements

Risk: May not work with all UIs

• Mitigation: Stick to OpenAI-compatible streaming format

Open Questions

1. Should we support function calling format for backwards compatibility?
2. How verbose should reasoning output be?
3. Should we cache tool results?
4. Do we need user confirmation for destructive operations?
5. Should tools have permission levels based on user?

---

Ready to implement: Yes ✓ Estimated timeline: 1-2 weeks Priority: High (enables true agentic behavior)