jpmschweitzer/portainer-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ai): complete ADK migration and optimize system health checks

Browse Source 
Major architectural changes and improvements:

## ADK Framework Migration (v0.10.0)
- Migrated from LangChain/LangGraph to Google ADK 1.3.0 with LiteLLM 1.80.5
- Improved tool calling reliability with local Ollama models
- Converted all 10 tools to ADK async generator format
- Updated streaming pipeline for ADK event system
- Enhanced error handling and agent initialization

## Model Optimization
- Switched from gemma3:12b (10GB VRAM) to gemma3:4b (4.8GB VRAM)
- Reduced VRAM usage from 91% to 43% (5.4GB freed)
- Optimized for production stability with memory headroom

## Health Check System Overhaul
- Optimized /health/full: 6ms response (was 30s+)
- Added model verification: confirms configured model is available
- New /health/diagnostics endpoint with optional deep testing
- Added currently loaded models tracking
- Clear emoji status indicators (✅/❌/⚠️)
- Fixed AGENT_AVAILABLE flag export for proper health reporting

## Ollama Client Enhancements
- Added list_models() method for model inventory
- Enhanced model verification in health checks
- Better error handling and reporting

## Documentation Updates
- Updated STATUS.md to v0.10.0-adk-migration
- Comprehensive CHANGELOG.md entry with migration details
- Updated PLANS.md showing Phase 4 complete
- Updated ai-orchestrator-plan.md with ADK status
- Added MIGRATION_PLAN_LANGCHAIN_TO_ADK.md
- Added ADK_Ollama_Research.md with implementation analysis

## Technical Details
- 10 tools: 7 infrastructure + 2 research + 1 response tool
- Framework: Google ADK with UnifiedAgent pattern
- System prompt: v7_adk_best_practice
- Container health: Now passing Docker healthchecks
- Response times: Simple queries ~0.3-1s, Research ~4-7s
This commit is contained in:
jpmschweitzer
2025-11-26 08:36:50 +01:00
parent bfc58b03ba
commit e3b451b7b0
15 changed files with 1731 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/core-api/src/agent/__init__.py
+14 -2
View File
@@ -4,12 +4,24 @@ Unified Agent Module
This module provides an intelligent agent that can handle infrastructure management,
web search, and multi-step reasoning with transparent streaming output.
"""
from .orchestrator import UnifiedAgent, get_unified_agent
from .tools import get_agent_tools, ALL_TOOLS
# Check if agent dependencies are available
try:
from .orchestrator import UnifiedAgent, get_unified_agent
from .tools import get_agent_tools, ALL_TOOLS
AGENT_AVAILABLE = True
except ImportError as e:
# ADK or dependencies not available
AGENT_AVAILABLE = False
UnifiedAgent = None
get_unified_agent = None
get_agent_tools = None
ALL_TOOLS = []
__all__ = [
"UnifiedAgent",
"get_unified_agent",
"get_agent_tools",
"ALL_TOOLS",
"AGENT_AVAILABLE",
]
services/core-api/src/agent/orchestrator.py
+163 -113
View File
@@ -1,90 +1,102 @@
"""
Agent Orchestrator - Unified intelligent agent with streaming reasoning
Agent Orchestrator - Unified intelligent agent with streaming reasoning using Google ADK
This orchestrator uses LangGraph to create a ReAct-style agent that can:
This orchestrator uses Google's Agent Development Kit (ADK) to create an agent that can:
- Use tools to answer infrastructure questions
- Stream thinking/reasoning output
- Handle multi-step tasks
- Route to appropriate expert models
- Work with local Ollama models via LiteLLM
"""
import json
import os
import logging
from typing import AsyncIterator, Dict, Any, List
from typing import AsyncIterator, Dict, Any, List, Optional
from functools import lru_cache
from langchain_ollama import ChatOllama
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
try:
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
ADK_AVAILABLE = True
except ImportError:
ADK_AVAILABLE = False
logger_temp = logging.getLogger(__name__)
logger_temp.error("Google ADK not installed! Run: pip install google-adk litellm")
from src.config import get_settings
from src.agent.tools import get_agent_tools
from src.agent.prompts import get_prompt
import uuid
logger = logging.getLogger(__name__)
class UnifiedAgent:
"""
Unified intelligent agent that handles all tool routing and reasoning
Unified intelligent agent that handles all tool routing and reasoning using Google ADK
"""
def __init__(self):
if not ADK_AVAILABLE:
raise ImportError("Google ADK is not installed. Please install: pip install google-adk")
self.settings = get_settings()
self.tools = get_agent_tools()
# Initialize Ollama LLM (must be a model that supports tool calling)
self.llm = ChatOllama(
model=self.settings.agent_model,
base_url=self.settings.ollama_base_url,
temperature=0.7,
# Initialize LiteLLM for Ollama (format: "ollama/model_name")
model_name = self.settings.agent_model
litellm_model = f"ollama/{model_name}"
logger.info(f"Initializing ADK with LiteLLM model: {litellm_model}")
logger.info(f"Ollama base URL from settings: {self.settings.ollama_base_url}")
self.llm = LiteLlm(
model=litellm_model,
api_base=self.settings.ollama_base_url,
response_format={"type": "text"},
force_json=False,
temperature=0.1,
)
# Create ReAct agent with tools
self.agent = create_react_agent(
self.llm,
self.tools,
state_modifier=self._get_system_prompt(),
# Store agents with different prompts for A/B testing
self._agents = {}
# Create default agent
self.agent = self._get_agent(self.settings.system_prompt_variant)
# Create session service and runner for executing the agent
session_service = InMemorySessionService()
self.runner = Runner(
app_name="portainer-core-api",
agent=self.agent,
session_service=session_service
)
logger.info(f"Initialized Unified Agent with {len(self.tools)} tools")
logger.info(f"Initialized ADK Agent with {len(self.tools)} tools using prompt variant: {self.settings.system_prompt_variant}")
def _get_system_prompt(self) -> str:
"""Get the system prompt that defines agent behavior"""
return """Your name is Tatlock, a helpful personal assistant with the demeanor of a British butler.
You address users as \"sir\" and speak formally.
You are not overly apologetic and can be a little snarky at times.
def _get_agent(self, prompt_variant: str) -> Agent:
"""Get or create an agent with a specific prompt variant"""
if prompt_variant not in self._agents:
system_prompt = get_prompt(prompt_variant)
Your capabilities:
- Search the web and extract content
- Monitor service health via Uptime Kuma
- Read project documentation
- Check system resources
- Manage Docker containers and services via Portainer
- Configure reverse proxies and domains via Nginx Proxy Manager
self._agents[prompt_variant] = Agent(
model=self.llm,
name="tatlock",
description="British butler assistant for technical household matters",
instruction=system_prompt,
tools=self.tools,
)
logger.info(f"Created ADK agent with prompt variant: {prompt_variant}")
When helping users:
1. Think step-by-step about what information you need
2. Use tools when you need current/specific information
3. Be concise but thorough in your responses
4. If a task requires multiple steps, explain what you're doing
5. Always verify information before making changes
6. Use lists and tables to display structured responses
7. Visualize statistics if clear categories and trendis are occurring
Available infrastructure:
- 22 running services (Ollama, Portainer, NPM, Jellyfin, Gitea, etc.)
- GPU: NVIDIA RTX 2080 Ti (11GB VRAM)
- Storage: SSD for configs, HDD for media
- Network: Headscale mesh VPN + NPM reverse proxy
If you see an opportunity to make a pun or joke, you simply cannot resist.
Be helpful, accurate, and transparent about what you're doing!"""
return self._agents[prompt_variant]
async def chat(
self,
message: str,
conversation_history: List[Dict[str, str]] = None,
stream: bool = True
stream: bool = True,
prompt_variant: Optional[str] = None
) -> AsyncIterator[Dict[str, Any]]:
"""
Process a chat message with streaming reasoning output
@@ -93,85 +105,121 @@ Be helpful, accurate, and transparent about what you're doing!"""
message: User's message
conversation_history: Previous conversation turns (optional)
stream: Whether to stream intermediate steps
prompt_variant: Override system prompt variant for A/B testing (optional)
Yields:
Dict with keys:
- type: "thinking" | "tool_call" | "tool_result" | "content"
- type: "thinking" | "tool_call" | "tool_result" | "content" | "error"
- content: The actual content
- tool: Tool name (if type is tool_call)
- model: Model being used (optional)
"""
try:
# Build message list
messages = []
# Generate session ID (use conversation history to maintain session)
# For now, create a new session per request (stateless)
user_id = "default_user"
session_id = str(uuid.uuid4())
# Add conversation history if provided
if conversation_history:
for turn in conversation_history:
if turn.get("role") == "user":
messages.append(HumanMessage(content=turn["content"]))
elif turn.get("role") == "assistant":
messages.append(AIMessage(content=turn["content"]))
# Create session
await self.runner.session_service.create_session(
app_name="portainer-core-api",
user_id=user_id,
session_id=session_id
)
# Add current message
messages.append(HumanMessage(content=message))
# Create Content object from message
new_message = Content(
parts=[Part(text=message)],
role="user"
)
# Initial thinking
yield {
"type": "thinking",
"content": "Analyzing your request...",
"model": self.settings.default_model
}
logger.info(f"🚀 Starting ADK Runner with message 🧠: {message[:50]}...")
# Stream agent execution
async for chunk in self.agent.astream(
{"messages": messages},
stream_mode="values" # Stream full state updates
# Track if we've seen any content
has_content = False
logger.info("About to start async iteration over runner.run_async()")
# Stream from ADK Runner
async for event in self.runner.run_async(
user_id=user_id,
session_id=session_id,
new_message=new_message
):
# Extract messages from the chunk
if "messages" in chunk:
latest_messages = chunk["messages"]
event_type_name = type(event).__name__
logger.info(f"ADK Event: {event_type_name}")
# Process the latest message
if latest_messages:
latest = latest_messages[-1]
# SPECIAL HANDLING for the model hallucinating a 'response' tool call.
# The model sometimes calls `response(answer=...)` for its final output,
# even when the prompt directs it not to. This intercepts that specific
# tool call and treats its input as the final content.
if event_type_name == "ToolCallStart" and event.tool_name == "response":
try:
answer = event.tool_input.get("answer", "")
if answer:
logger.info("📢 Intercepted 'response' tool. Delivering final answer.")
yield {"type": "content", "content": answer}
has_content = True
# Gracefully exit the generator as this is the final response.
return
except Exception as e:
logger.error(f"Error processing special 'response' tool call: {e}")
# Tool invocation
if hasattr(latest, 'additional_kwargs') and 'tool_calls' in latest.additional_kwargs:
tool_calls = latest.additional_kwargs['tool_calls']
for tool_call in tool_calls:
tool_name = tool_call.get('function', {}).get('name', 'unknown')
yield {
"type": "tool_call",
"tool": tool_name,
"content": f"Using tool: {tool_name}..."
}
# Map ADK events to our format
event_type = type(event).__name__
# Tool result
elif isinstance(latest, ToolMessage):
yield {
"type": "tool_result",
"content": "Tool execution complete"
}
if event_type == "ToolCallStart":
# Tool is being called
tool_name = getattr(event, "tool_name", "unknown")
logger.info(f"🔧 TOOL CALL START: {tool_name}")
# AI response (final or intermediate)
elif isinstance(latest, AIMessage) and latest.content:
# Check if this is intermediate thinking or final response
if hasattr(latest, 'additional_kwargs') and latest.additional_kwargs.get('tool_calls'):
# This is thinking before a tool call
yield {
"type": "thinking",
"content": latest.content
}
else:
# This is the final response
yield {
"type": "content",
"content": latest.content
}
yield {
"type": "tool_call",
"tool": tool_name,
"content": f"Using tool: {tool_name}..."
}
elif event_type == "ToolCallEnd":
# Tool execution completed
tool_name = getattr(event, "tool_name", "unknown")
logger.info(f"✅ TOOL CALL END: {tool_name}")
yield {
"type": "tool_result",
"content": "Tool execution complete"
}
elif event_type == "ContentDelta":
# Stream content tokens
content = getattr(event, "content", "")
if content:
has_content = True
yield {
"type": "content",
"content": content
}
elif event_type == "AgentThinking":
# Agent reasoning (if available)
thinking = getattr(event, "content", "")
if thinking:
yield {
"type": "thinking",
"content": thinking
}
logger.info(f"✅ ADK agent stream completed (has_content={has_content})")
# If no content was yielded, provide a default message
if not has_content:
logger.warning("No content generated by agent")
yield {
"type": "content",
"content": "I apologize, but I wasn't able to generate a response."
}
except Exception as e:
logger.error(f"Error in agent chat: {e}", exc_info=True)
logger.error(f"Error in ADK agent chat: {e}", exc_info=True)
yield {
"type": "error",
"content": f"Sorry, I encountered an error: {str(e)}"
@@ -180,7 +228,8 @@ Be helpful, accurate, and transparent about what you're doing!"""
async def chat_completion(
self,
message: str,
conversation_history: List[Dict[str, str]] = None
conversation_history: List[Dict[str, str]] = None,
prompt_variant: Optional[str] = None
) -> str:
"""
Get a non-streaming response (for backwards compatibility)
@@ -188,12 +237,13 @@ Be helpful, accurate, and transparent about what you're doing!"""
Args:
message: User's message
conversation_history: Previous conversation turns (optional)
prompt_variant: Override system prompt variant for A/B testing (optional)
Returns:
The final response content
"""
final_content = ""
async for chunk in self.chat(message, conversation_history, stream=True):
async for chunk in self.chat(message, conversation_history, stream=True, prompt_variant=prompt_variant):
if chunk["type"] == "content":
final_content += chunk["content"]
services/core-api/src/agent/streaming.py
+34 -20
View File
@@ -55,8 +55,24 @@ async def stream_agent_to_sse(agent_stream: AsyncIterator[Dict[str, Any]], reque
yield f"data: {json.dumps(sse_chunk)}\n\n"
elif chunk_type == "tool_call":
# Stream tool call notification
# Stream tool call notification with enhanced icons for research
tool_name = chunk.get("tool", "unknown")
# Enhanced progress indicators for different tool types
tool_icons = {
"web_search": "🔍 Searching web",
"web_scrape": "📄 Reading page",
"list_services": "🔧 Listing services",
"get_service_details": "🔍 Checking service",
"list_domains": "🌐 Listing domains",
"check_service_health": "💚 Checking health",
"get_system_status": "📊 Getting system status",
"get_current_time": "🕐 Checking time",
"read_documentation": "📖 Reading docs"
}
display_text = tool_icons.get(tool_name, f"🔧 Using {tool_name}")
sse_chunk = {
"id": request_id,
"object": "chat.completion.chunk",
@@ -66,7 +82,7 @@ async def stream_agent_to_sse(agent_stream: AsyncIterator[Dict[str, Any]], reque
"index": 0,
"delta": {
"role": "assistant",
"content": f"[🔧 Using {tool_name}...]\n"
"content": f"[{display_text}...]\n"
},
"finish_reason": None
}]
@@ -93,24 +109,22 @@ async def stream_agent_to_sse(agent_stream: AsyncIterator[Dict[str, Any]], reque
elif chunk_type == "content":
# Stream actual content (final response)
# Split into words for smooth streaming
words = content.split()
for word in words:
sse_chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"delta": {
"content": word + " "
},
"finish_reason": None
}]
}
yield f"data: {json.dumps(sse_chunk)}\n\n"
chunk_index += 1
# Content is already token-level from orchestrator, just pass through
sse_chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{
"index": 0,
"delta": {
"content": content # Already tokenized, preserves formatting
},
"finish_reason": None
}]
}
yield f"data: {json.dumps(sse_chunk)}\n\n"
chunk_index += 1
elif chunk_type == "error":
# Stream error
services/core-api/src/agent/tools.py
+199 -22
View File
@@ -1,20 +1,48 @@
"""
Agent Tools - LangChain-compatible tools for the unified agent
Agent Tools - Google ADK-compatible tools for the unified agent
These tools wrap existing Core API functionality for use with LangGraph.
These tools wrap existing Core API functionality for use with ADK agents.
"""
from langchain_core.tools import tool
from typing import List, Dict, Optional
import logging
import functools
import inspect
logger = logging.getLogger(__name__)
def log_tool_call(func):
"""Decorator to log tool calls with their parameters"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
# Get function signature
sig = inspect.signature(func)
bound_args = sig.bind(*args, **kwargs)
bound_args.apply_defaults()
# Format parameters for logging
params_str = ", ".join(f"{k}={repr(v)}" for k, v in bound_args.arguments.items())
logger.info(f"🔧 TOOL CALL: {func.__name__}({params_str})")
try:
result = await func(*args, **kwargs)
# Log result preview (first 200 chars)
result_preview = str(result)[:200] if result else "None"
logger.info(f"✅ TOOL RESULT: {func.__name__} → {result_preview}...")
return result
except Exception as e:
logger.error(f"❌ TOOL ERROR: {func.__name__} failed with {type(e).__name__}: {e}")
raise
return wrapper
# ============================================================================
# Infrastructure Management Tools
# ============================================================================
@tool
@log_tool_call
async def list_services() -> str:
"""
List all running Docker services on the homelab server.
@@ -53,7 +81,7 @@ async def list_services() -> str:
return f"Error: Could not list services - {str(e)}"
@tool
@log_tool_call
async def get_service_details(service_name: str) -> str:
"""
Get detailed information about a specific Docker service.
@@ -88,7 +116,8 @@ async def get_service_details(service_name: str) -> str:
return f"Error: Could not get details for '{service_name}' - {str(e)}"
@tool
# @tool - removed for ADK
@log_tool_call
async def list_domains() -> str:
"""
List all configured domain names and their proxy configurations.
@@ -124,7 +153,8 @@ async def list_domains() -> str:
return f"Error: Could not list domains - {str(e)}"
@tool
# @tool - removed for ADK
@log_tool_call
async def check_service_health(service_name: str) -> str:
"""
Check the health status of a service via Uptime Kuma monitoring.
@@ -150,13 +180,86 @@ async def check_service_health(service_name: str) -> str:
# Knowledge & Search Tools
# ============================================================================
@tool
async def web_search(url: str) -> str:
# @tool - removed for ADK
@log_tool_call
async def web_search(query: str, num_results: int) -> str:
"""
Fetch and extract the main content from a web page.
Search the web using DuckDuckGo and extract content from top results.
Uses DuckDuckGo to find relevant web pages, then extracts the main content from each result.
Perfect for answering questions that require current information from the web.
Args:
query: The search query (e.g., "LangGraph documentation", "latest news about AI")
num_results: Number of results to return (max 5)
Returns:
Formatted search results with titles, URLs, snippets, and extracted content
"""
try:
from duckduckgo_search import DDGS
from src.web_scraper.service import WebScraperService
scraper = WebScraperService()
num_results = min(num_results, 5) # Cap at 5 results
results = []
with DDGS() as ddgs:
search_results = list(ddgs.text(query, max_results=num_results))
if not search_results:
return f"No search results found for: {query}"
for idx, result in enumerate(search_results, 1):
title = result.get('title', 'Unknown')
url = result.get('href', '')
snippet = result.get('body', '')
# Try to scrape content from the page
content = ""
try:
scrape_result = await scraper.scrape_url(url)
if scrape_result and scrape_result.content:
# Get first 500 chars of content
content = scrape_result.content[:500]
if len(scrape_result.content) > 500:
content += "..."
except Exception as scrape_error:
logger.warning(f"Could not scrape {url}: {scrape_error}")
content = snippet # Fall back to snippet
results.append({
'index': idx,
'title': title,
'url': url,
'snippet': snippet,
'content': content
})
# Format results for LLM
output = f"Search results for '{query}':\n\n"
for r in results:
output += f"{r['index']}. **{r['title']}**\n"
output += f" URL: {r['url']}\n"
output += f" {r['content']}\n\n"
output += "\nNote: Synthesize information from these sources and cite URLs in your response."
return output
except Exception as e:
logger.error(f"Error performing web search: {e}")
return f"Error: Could not search the web - {str(e)}"
# @tool - removed for ADK
@log_tool_call
async def web_scrape(url: str) -> str:
"""
Fetch and extract the main content from a specific web page.
Uses intelligent content extraction to get the most relevant text from articles,
documentation, and blog posts. Perfect for answering questions that require current information.
documentation, and blog posts. Use this when you have a specific URL to read.
Args:
url: The URL to fetch and extract content from
@@ -185,7 +288,8 @@ async def web_search(url: str) -> str:
return f"Error: Could not fetch content from {url} - {str(e)}"
@tool
# @tool - removed for ADK
@log_tool_call
async def read_documentation(topic: str) -> str:
"""
Read project documentation files.
@@ -222,7 +326,40 @@ async def read_documentation(topic: str) -> str:
# System Information Tools
# ============================================================================
@tool
# @tool - removed for ADK
@log_tool_call
async def get_current_time() -> str:
"""
Get the current date and time.
Use this when the user asks about the current time, date, or when you need
to know "today's date" for searches (e.g., "today's news", "today's stock prices").
Returns:
Current date and time in a readable format
"""
from datetime import datetime
import pytz
try:
# Get current time in UTC and local timezone
utc_now = datetime.now(pytz.UTC)
# Central European Time (Amsterdam/Netherlands)
local_tz = pytz.timezone('Europe/Amsterdam')
local_now = utc_now.astimezone(local_tz)
result = f"Current Time:\n"
result += f"Local: {local_now.strftime('%A, %B %d, %Y at %H:%M %Z')}\n"
result += f"UTC: {utc_now.strftime('%A, %B %d, %Y at %H:%M %Z')}\n"
return result
except Exception as e:
logger.error(f"Error getting current time: {e}")
return f"Error: Could not get current time - {str(e)}"
# @tool - removed for ADK
@log_tool_call
async def get_system_status() -> str:
"""
Get current system status including resource usage.
@@ -262,18 +399,58 @@ async def get_system_status() -> str:
# ============================================================================
# Tool Registry
# Special Tools
# ============================================================================
# All available tools for the agent
@log_tool_call
async def response(answer: str) -> str:
"""
Deliver your final response to the user.
Use this tool ONLY when you want to provide your final answer to the user after gathering
information from other tools. This is your way of speaking directly to the user.
Args:
answer: Your complete response to the user in natural language
Returns:
Confirmation that the response was delivered
"""
# This is a special tool - it just returns the answer back
# The orchestrator will recognize this and end the conversation
return answer
# ============================================================================
# Tool Registry - ADK Format
# ============================================================================
# Import ADK FunctionTool for wrapping
try:
from google.adk.tools import FunctionTool
ADK_AVAILABLE = True
except ImportError:
# Fallback if ADK not installed yet
ADK_AVAILABLE = False
FunctionTool = None
# All available tools for the agent (ADK FunctionTool format)
# ADK's FunctionTool extracts name and description from the function itself
ALL_TOOLS = [
list_services,
get_service_details,
list_domains,
check_service_health,
web_search,
read_documentation,
get_system_status,
# Infrastructure tools
FunctionTool(list_services),
FunctionTool(get_service_details),
FunctionTool(list_domains),
FunctionTool(check_service_health),
# Knowledge & search tools
FunctionTool(web_search),
FunctionTool(web_scrape),
FunctionTool(read_documentation),
# System information tools
FunctionTool(get_current_time),
FunctionTool(get_system_status),
# Special response tool
FunctionTool(response),
]