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

refactor(core-ai): comprehensive cleanup - PydanticAI only architecture

Browse Source 
Remove all obsolete agent implementations and framework references.
Keep only PydanticAI (primary) and SimpleLiteLLM (fallback).

This cleanup eliminates confusion between multiple frameworks that were
tried during development (LangChain, LangGraph, ADK, OllamaNative) and
establishes PydanticAI as the single agent framework going forward.

BREAKING CHANGES:
- Removed OllamaNativeAgent - use PydanticAgent instead
- Removed /test/ollama-tools diagnostic endpoint
- Default /v1/chat/completions now uses PydanticAgent

Files Deleted (32 total):
- Obsolete agents: ollama_native_agent.py
- Diagnostic files: ARCHITECTURE.md, DIAGNOSTIC_RESULTS.md, PHASE*.md
- Legacy tools: src/tools.py
- Test files: test_ai_flow_quality.py, test_02/03 (diagnostic layers)
- Documentation: ADK_Ollama_Research.md, agent-flow-diagrams.md
- Session docs: 3 files with LangChain/LangGraph implementations
- Plans: 5 completed plans about obsolete frameworks
- Migration docs: MIGRATION_PLAN_LANGCHAIN_TO_ADK.md

Files Modified (8 total):
- main.py: Refactored to PydanticAI only (305 lines vs 457 before)
- agents/__init__.py: Removed OllamaNativeAgent exports
- README.md: Complete rewrite for PydanticAI architecture
- prompts.py: Updated for PydanticAI (infrastructure tool guidance)
- STATUS.md: Updated to v0.11.0-pydantic-ai
- CHANGELOG.md: Added v0.11.0 entry documenting cleanup
- plans/active/*.md: Updated to reference PydanticAI

Current Architecture:
- Framework: PydanticAI with native Ollama SDK
- Agents: PydanticAgent (primary) + SimpleLiteLLMAgent (fallback)
- Model: mistral-nemo:latest
- Tools: 6 core + 28+ OpenAPI-discovered
- Memory: 3-tier system with Qdrant
- VRAM: ~4-6GB

Lines Removed: ~3000+ lines of obsolete code

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2025-12-03 14:24:52 +01:00
co-authored by Claude
parent 96492cb1ed
commit 66f6e54fc3
32 changed files with 1078 additions and 10376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/core-ai/ARCHITECTURE.md
-334
View File
@@ -1,334 +0,0 @@
# Core-AI Dual-Mode Architecture
## Overview
Core-AI provides **two AI endpoints** with different complexity levels:
1. **Simple Mode** - Direct LiteLLM (existing)
2. **ADK Mode** - Full Google ADK with tool calling (new)
Core-API becomes a **pure tools platform** providing REST endpoints.
---
## Architecture Diagram
```
┌─────────────────────────────────────────────────────────────┐
│ Core-AI │
│ │
│ ┌─────────────────────┐ ┌─────────────────────┐ │
│ │ Simple Endpoint │ │ ADK Endpoint │ │
│ │ /v1/chat/simple │ │ /v1/chat/adk │ │
│ │ │ │ │ │
│ │ SimpleLiteLLMAgent │ │ ADKAgent │ │
│ │ ↓ │ │ ↓ │ │
│ │ LiteLLM │ │ ADK Runtime │ │
│ │ ↓ │ │ ↓ │ │
│ │ [No Tools] │ │ Tool Registry │ │
│ └─────────────────────┘ │ ↓ │ │
│ │ REST Calls ───────┼─────┼─┐
│ └─────────────────────┘ │ │
└───────────────────────────────────────────────────────────┘ │
│
┌────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Core-API │
│ (Tools Platform) │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ System │ │ Services │ │ Docker │ │
│ │ Tools │ │ Tools │ │ Tools │ │
│ │ /system/* │ │ /services/* │ │ /docker/* │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ [No AI Agent - Pure REST API] │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ External Systems │
│ (Portainer, Uptime Kuma, Ollama, etc.) │
└─────────────────────────────────────────────────────────────┘
```
---
## API Routes
### Core-AI Routes
| Endpoint | Mode | Agent | Tools | Use Case |
|----------|------|-------|-------|----------|
| `POST /v1/chat/simple` | Simple | SimpleLiteLLMAgent | None | Fast Q&A, text generation |
| `POST /v1/chat/adk` | ADK | ADKAgent | Full toolset | Complex tasks, orchestration |
| `GET /health` | N/A | N/A | N/A | Health check |
### Core-API Routes (No Changes - Tools Only)
| Endpoint | Purpose |
|----------|---------|
| `GET /v1/system/status` | System status |
| `GET /v1/services/*` | Service management |
| `GET /v1/docker/*` | Docker operations |
| `POST /v1/tools/*` | Tool execution |
---
## Request/Response Formats
### Simple Mode
```json
POST /v1/chat/simple
{
"messages": [
{"role": "user", "content": "What is 2+2?"}
],
"stream": false
}
Response:
{
"id": "chatcmpl-...",
"object": "chat.completion",
"model": "simple",
"choices": [{
"message": {"role": "assistant", "content": "4"},
"finish_reason": "stop"
}]
}
```
### ADK Mode
```json
POST /v1/chat/adk
{
"messages": [
{"role": "user", "content": "Check the system status and tell me if everything is ok"}
],
"stream": true
}
Response (streaming):
data: {"type": "tool_call", "tool": "get_system_status", ...}
data: {"type": "tool_result", "result": {...}}
data: {"type": "content", "content": "Everything looks good..."}
data: {"type": "content", "finish_reason": "stop"}
```
---
## File Structure
```
services/core-ai/
├── main.py # HTTP server with both routes
├── src/
│ ├── __init__.py
│ ├── config.py # Configuration
│ ├── prompts.py # System prompts (simple + ADK)
│ ├── agents/
│ │ ├── __init__.py
│ │ ├── simple.py # SimpleLiteLLMAgent (existing)
│ │ └── adk_agent.py # ADKAgent (new)
│ └── tools/
│ ├── __init__.py
│ ├── registry.py # ADK tool registry
│ ├── system_tools.py # System tools (REST calls to core-api)
│ ├── service_tools.py # Service tools (REST calls to core-api)
│ └── knowledge_tools.py # Knowledge tools (web search, etc.)
├── diagnostics/
│ ├── check_ollama.py
│ ├── test_litellm_direct.py
│ └── test_adk_direct.py # New: Test ADK without HTTP
├── tests/
│ ├── test_01_environment.py # ✓ Existing
│ ├── test_02_litellm_raw.py # ✓ Existing
│ ├── test_03_message_format.py # ✓ Existing
│ ├── test_04_agent.py # ✓ Existing (simple agent)
│ ├── test_05_api.py # ✓ Existing (simple API)
│ ├── test_06_adk_setup.py # New: ADK initialization
│ ├── test_07_adk_tools.py # New: ADK tool registration
│ ├── test_08_adk_agent.py # New: ADK agent logic
│ ├── test_09_adk_tool_calling.py # New: ADK tool execution
│ ├── test_10_adk_api.py # New: ADK API endpoint
│ ├── run_all_tests.sh
│ └── README.md
└── README.md
```
---
## Implementation Phases
### Phase 1: ADK Agent Setup ✓
- [x] Create `src/agents/adk_agent.py`
- [x] Initialize ADK runtime
- [x] Test basic ADK completion
- [x] Create diagnostic: `diagnostics/test_adk_direct.py`
- [x] Create test: `tests/test_06_adk_setup.py`
### Phase 2: Tool Integration
- [ ] Create tool registry with ADK FunctionTool format
- [ ] Implement REST-based tools (call core-api endpoints)
- [ ] Test tool registration
- [ ] Create test: `tests/test_07_adk_tools.py`
### Phase 3: ADK Agent with Tools
- [ ] Integrate tools into ADK agent
- [ ] Test tool calling flow
- [ ] Verify REST calls to core-api
- [ ] Create tests: `test_08_adk_agent.py`, `test_09_adk_tool_calling.py`
### Phase 4: API Routes
- [ ] Add `/v1/chat/adk` endpoint
- [ ] Rename existing to `/v1/chat/simple` (keep `/v1/chat/completions` as alias)
- [ ] Test both endpoints
- [ ] Create test: `tests/test_10_adk_api.py`
### Phase 5: Documentation & Cleanup
- [ ] Update README.md
- [ ] Update test documentation
- [ ] Add architecture diagrams
- [ ] Document migration from core-api
---
## Tool Design: REST-First
All tools in core-ai make REST calls to core-api:
```python
# Example: System Status Tool
@log_tool_call
async def get_system_status() -> str:
"""Get current system status from core-api."""
async with httpx.AsyncClient() as client:
response = await client.get(f"{CORE_API_BASE_URL}/system/status")
data = response.json()
return json.dumps(data, indent=2)
```
**Benefits:**
- Clean separation: core-ai = AI, core-api = tools
- Tools can be used by both AI and direct API calls
- Easy to test tools independently
- No code duplication
---
## Testing Strategy
### Layer 1-5: Simple Mode (Existing)
Already tested and passing ✓
### Layer 6: ADK Setup
```python
# Test ADK runtime initialization
# Test ADK basic completion (no tools)
# Test ADK message handling
```
### Layer 7: ADK Tools
```python
# Test tool registration
# Test tool discovery
# Test REST connectivity to core-api
```
### Layer 8: ADK Agent
```python
# Test agent with tools
# Test prompt handling
# Test error handling
```
### Layer 9: ADK Tool Calling
```python
# Test tool invocation
# Test tool results
# Test multi-tool workflows
```
### Layer 10: ADK API
```python
# Test /v1/chat/adk endpoint
# Test streaming with tools
# Test non-streaming with tools
```
---
## Configuration
### Environment Variables
```bash
# Existing
OLLAMA_BASE_URL=http://ollama:11434
AGENT_MODEL=gemma2:9b-instruct-q5_K_M
SYSTEM_PROMPT_VARIANT=minimal_agent
HOST=0.0.0.0
PORT=8086
# New
CORE_API_BASE_URL=http://core-api:8083/v1 # For tool REST calls
ADK_ENABLED=true # Enable ADK endpoint
SIMPLE_ENABLED=true # Enable simple endpoint
ADK_SYSTEM_PROMPT_VARIANT=adk_agent # Different prompt for ADK
```
### Prompts
```python
PROMPTS = {
"minimal_agent": "You are a helpful assistant.", # Simple mode
"adk_agent": """You are a system management assistant with access to tools.
Use tools when needed to answer questions about system status, services, and docker."""
}
```
---
## Migration Path (Core-API)
**Later cleanup - not in this phase:**
1. Remove AI agent code from core-api
2. Remove ADK dependencies from core-api
3. Keep only REST endpoints
4. Update core-api to be pure API
5. Redirect any AI requests to core-ai
---
## Backward Compatibility
- `/v1/chat/completions` → alias for `/v1/chat/simple`
- Existing clients keep working
- New clients can choose mode
---
## Performance Considerations
| Aspect | Simple Mode | ADK Mode |
|--------|-------------|----------|
| **Latency** | ~0.5-1s | ~1-3s (with tools) |
| **Overhead** | Minimal | ADK runtime |
| **Memory** | Low | Medium (tool registry) |
| **Use Case** | Fast Q&A | Complex tasks |
---
## Next Steps
1. ✅ Design architecture (this document)
2. ⏳ Implement Phase 1: ADK Agent Setup
3. ⏳ Implement Phase 2: Tool Integration
4. ⏳ Implement Phase 3: ADK Agent with Tools
5. ⏳ Implement Phase 4: API Routes
6. ⏳ Implement Phase 5: Documentation
**Let's start with Phase 1!**
services/core-ai/DIAGNOSTIC_RESULTS.md
-330
View File
@@ -1,330 +0,0 @@
# Core-AI Diagnostic Results
**Date:** 2025-11-27
**Status:** ✅ ALL SYSTEMS OPERATIONAL
## Executive Summary
The core-ai service **IS WORKING CORRECTLY** and can successfully answer simple questions like "What is the capital of France?"
The investigation revealed that the basic LiteLLM → Ollama → Model stack was functional, but **lacked proper diagnostics and logging** to identify issues when they occur. We've now added comprehensive testing and improved observability.
---
## Test Results
### ✅ Ollama Connectivity Check
```
Status: PASSED
- Ollama is reachable at http://ollama:11434
- Target model 'gemma2:9b-instruct-q5_K_M' is available (6.19 GB)
- Text generation test successful
```
### ✅ Direct LiteLLM Tests
```
Status: ALL 3 TESTS PASSED
Test 1: Simple question (no system prompt)
Non-streaming: ✓ "Paris"
Streaming: ✓ "Paris" (4 chunks)
Test 2: Simple question (with system prompt)
Non-streaming: ✓ "Paris"
Streaming: ✓ "Paris" (4 chunks)
Test 3: Math problem
Non-streaming: ✓ "4"
Streaming: ✓ "4" (2 chunks)
```
### ✅ End-to-End API Test
```bash
$ curl -X POST http://localhost:8086/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{"messages": [{"role": "user", "content": "What is the capital of France?"}]}'
Response: "The capital of France is Paris."
Status: 200 OK
```
---
## Issues Found and Fixed
### 1. Configuration Mismatch ⚠️ FIXED
**Location:** `stacks/core-ai.yml:18`
**Problem:**
```yaml
SYSTEM_PROMPT_VARIANT=v8_holistic # ❌ This variant doesn't exist
```
**Fix:**
```yaml
SYSTEM_PROMPT_VARIANT=minimal_agent # ✅ Matches prompts.py
```
**Impact:** Low - Service would use fallback prompt anyway, but could cause confusion.
---
### 2. Missing System Prompt Integration ⚠️ FIXED
**Location:** `services/core-ai/src/agent.py`
**Problem:** Agent wasn't injecting system prompt into messages before sending to LiteLLM.
**Fix:** Added:
- System prompt loading in `__init__()`
- System prompt injection logic in `chat()`
- Logging of system prompt and full message payload
**Impact:** Medium - Without system prompt, model behavior could be unpredictable.
---
### 3. Insufficient Diagnostics ⚠️ FIXED
**Problem:** No way to systematically test each component.
**Fix:** Created comprehensive test suite:
- Layer 1: Environment & Configuration tests
- Layer 2: Raw LiteLLM connection tests
- Layer 3: Message formatting tests
- Layer 4: Agent logic tests
- Layer 5: API integration tests
**Impact:** High - Previously couldn't pinpoint failure locations.
---
### 4. Poor Logging ⚠️ FIXED
**Problem:** Logs didn't show what was being sent to LiteLLM.
**Fix:** Added detailed logging:
- System prompt variant and content
- Full message payload with roles
- Response content and finish reasons
- Streaming chunk counts
**Impact:** High - Now can diagnose issues from logs alone.
---
## What Was Already Working
✅ **LiteLLM → Ollama Integration**
The core connection was solid from the start.
✅ **Model Selection**
gemma2:9b-instruct-q5_K_M was properly configured and loaded.
✅ **Basic Text Generation**
Model could generate responses to simple questions.
✅ **API Endpoints**
HTTP server, routing, and OpenAI-compatible format all functional.
---
## Root Cause Analysis
**Question:** Why did the user think the service couldn't answer "What is the capital of France?"
**Possible Reasons:**
1. **Previous Build Had Issues**
The service was working in the latest version, but may have had problems in an earlier iteration.
2. **Lack of Visibility**
Without diagnostics, it was hard to tell if the service was working or not.
3. **Configuration Confusion**
The `v8_holistic` prompt variant mismatch may have caused uncertainty.
4. **Testing from Wrong Context**
If tested from outside Docker network or with wrong endpoint, would appear broken.
---
## Current Service Health
### Response Times
- Simple questions: ~0.5-1s
- With system prompt: ~0.5-1s
- Streaming mode: Real-time chunks
### Accuracy
- ✅ "What is the capital of France?" → "Paris"
- ✅ "What is 2+2?" → "4"
- ✅ Follows system prompt instructions
- ✅ Handles both streaming and non-streaming
### Resource Usage
- Container: Running stable
- Model: Loaded in Ollama (6.19 GB)
- Memory: Within normal limits
- CPU: Minimal when idle
---
## Improvements Made
### 1. Enhanced Logging
```
2025-11-27 11:19:36 - INFO - System prompt variant: minimal_agent
2025-11-27 11:19:36 - INFO - System prompt: You are a helpful assistant...
2025-11-27 11:19:36 - INFO - ✓ System prompt injected
2025-11-27 11:19:36 - INFO - 📤 Sending 2 messages to LiteLLM:
2025-11-27 11:19:36 - INFO - [0] system: You are a helpful assistant...
2025-11-27 11:19:36 - INFO - [1] user: What is 2+2? Just the number.
2025-11-27 11:19:36 - INFO - 📥 Response received: 4
```
### 2. Diagnostic Tools
- `diagnostics/check_ollama.py` - Verify Ollama connectivity
- `diagnostics/test_litellm_direct.py` - Test raw LiteLLM integration
### 3. Test Suite
- 5 layers of tests (environment → API)
- Automated test runner (`tests/run_all_tests.sh`)
- Clear pass/fail indicators
- Stops at first failure for easy debugging
### 4. Documentation
- `README.md` - Service documentation
- `tests/README.md` - Testing guide
- `DIAGNOSTIC_RESULTS.md` - This file
---
## Next Steps
### Option 1: Keep Core-AI as Lean Service (Recommended)
**Use Case:** Simple text generation without ADK complexity
**Advantages:**
- ✅ Low overhead
- ✅ Easy to debug
- ✅ Fast response times
- ✅ Good for simple tasks
**When to use:**
- Basic Q&A
- Text completion
- Simple chat
- Testing Ollama models
### Option 2: Migrate Improvements to Core-API
**Use Case:** Production service with full ADK + tool calling
**Tasks:**
1. Apply logging improvements to core-api
2. Add system prompt injection verification
3. Port diagnostic tools
4. Create test suite for ADK layer
### Option 3: Keep Both (Hybrid Approach)
**Use Case:** Different services for different needs
**Architecture:**
```
┌─────────────┐ ┌──────────────┐
│ Core-AI │ │ Core-API │
│ (Simple) │ │ (Full ADK) │
└─────┬───────┘ └──────┬───────┘
│ │
└──────┬─────────────┘
│
┌────▼─────┐
│ LiteLLM │
└────┬─────┘
│
┌────▼─────┐
│ Ollama │
└────┬─────┘
│
┌────▼─────┐
│ Models │
└──────────┘
```
**Benefits:**
- Core-AI for simple, fast queries
- Core-API for complex orchestration
- Shared Ollama backend
- Different performance profiles
---
## Testing Checklist
To verify the service after any changes:
```bash
# 1. Check Ollama connectivity
docker exec core-ai python diagnostics/check_ollama.py
# 2. Test direct LiteLLM
docker exec core-ai python diagnostics/test_litellm_direct.py
# 3. Test end-to-end
curl -X POST http://localhost:8086/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{"messages": [{"role": "user", "content": "What is the capital of France?"}]}'
# 4. Check logs for detailed diagnostics
docker logs core-ai --tail 50
```
---
## Performance Baseline
| Metric | Value | Notes |
|--------|-------|-------|
| **First Response Time** | ~0.5-1s | Simple questions |
| **Streaming Latency** | Real-time | Chunks as available |
| **Model Load Time** | 0s | Already loaded |
| **Cold Start** | ~30s | First time pulling model |
| **Concurrent Requests** | Good | Limited by Ollama |
| **Memory per Request** | Minimal | Model stays loaded |
---
## Conclusion
The core-ai service is **fully functional** and correctly answers simple questions. The improvements made focus on **observability, diagnostics, and maintainability** rather than fixing broken functionality.
**Key Takeaway:** The foundation was solid; we added the tools to prove it and maintain it.
---
## Files Modified
### Configuration
- ✏️ `stacks/core-ai.yml` - Fixed SYSTEM_PROMPT_VARIANT
### Code
- ✏️ `services/core-ai/src/agent.py` - Added system prompt integration and logging
- ✏️ `services/core-ai/requirements.txt` - Added pytest dependencies
### New Files Created
- 📄 `services/core-ai/diagnostics/__init__.py`
- 📄 `services/core-ai/diagnostics/check_ollama.py`
- 📄 `services/core-ai/diagnostics/test_litellm_direct.py`
- 📄 `services/core-ai/tests/__init__.py`
- 📄 `services/core-ai/tests/test_01_environment.py`
- 📄 `services/core-ai/tests/test_02_litellm_raw.py`
- 📄 `services/core-ai/tests/test_03_message_format.py`
- 📄 `services/core-ai/tests/test_04_agent.py`
- 📄 `services/core-ai/tests/test_05_api.py`
- 📄 `services/core-ai/tests/run_all_tests.sh`
- 📄 `services/core-ai/tests/README.md`
- 📄 `services/core-ai/pytest.ini`
- 📄 `services/core-ai/README.md`
- 📄 `services/core-ai/DIAGNOSTIC_RESULTS.md` (this file)
---
**Last Updated:** 2025-11-27 12:20:00
**Test Status:** ✅ ALL PASSING
**Service Status:** ✅ OPERATIONAL
services/core-ai/PHASE1_COMPLETE.md
-269
View File
@@ -1,269 +0,0 @@
# Phase 1: ADK Agent Setup - COMPLETE ✓
**Date:** 2025-11-27
**Status:** Implementation Complete, Testing in Progress
---
## What Was Accomplished
### 1. Code Restructuring ✓
**Before:**
```
src/
├── agent.py # Single SimpleLiteLLMAgent
├── config.py
└── prompts.py
```
**After:**
```
src/
├── agents/
│ ├── __init__.py
│ ├── simple.py # SimpleLiteLLMAgent (moved)
│ └── adk_agent.py # ADKAgent (new)
├── config.py # Updated with ADK settings
└── prompts.py # Updated with ADK prompt
```
### 2. ADK Agent Implementation ✓
**File:** `src/agents/adk_agent.py`
**Features:**
- Google ADK integration with LiteLLM backend
- Streaming and non-streaming support
- Tool calling framework (ready for Phase 2)
- Event-driven architecture (tool_call, tool_result, content)
- Comprehensive logging
**API:**
```python
agent = ADKAgent(tools=[])
response = await agent.chat_completion(messages)
async for event in agent.chat(messages, stream=True):
# Handle events
```
### 3. Configuration Updates ✓
**File:** `src/config.py`
**New Settings:**
```python
adk_system_prompt_variant: str = "adk_agent" # Separate prompt for ADK
simple_enabled: bool = True # Feature flag
adk_enabled: bool = True # Feature flag
```
### 4. Prompt System ✓
**File:** `src/prompts.py`
**New Prompts:**
- `minimal_agent` - Simple mode (existing)
- `adk_agent` - ADK mode with tool guidance (new)
### 5. Diagnostic Tools ✓
**File:** `diagnostics/test_adk_direct.py`
**Tests:**
- ADK initialization
- Simple questions without tools
- Math problems
- Multi-step reasoning
- Streaming vs non-streaming
### 6. Test Suite Layer 6 ✓
**File:** `tests/test_06_adk_setup.py`
**Tests:**
- ADK import verification
- Prompt existence
- Agent initialization
- Simple completion
- Streaming mode
- "Capital of France" test
- System prompt loading
---
## Testing Strategy
### Phase 1 Tests (No Tools)
```bash
# Diagnostic test
docker exec core-ai python diagnostics/test_adk_direct.py
# Unit tests
docker exec core-ai pytest tests/test_06_adk_setup.py -v -s
```
### What We're Testing
✅ **ADK Runtime**
- Can import Google ADK
- Can initialize LiteLLM backend
- Can create ADK agent
✅ **Basic Completion**
- Simple questions work
- Math works
- Streaming works
- Non-streaming works
✅ **Configuration**
- Prompts load correctly
- Settings are applied
- Feature flags work
❌ **NOT Testing Yet (Phase 2)**
- Tool registration
- Tool calling
- REST integration
---
## Architecture
### Current State (Phase 1)
```
┌─────────────────────────────────────────┐
│ Core-AI Service │
│ │
│ ┌─────────────────┐ │
│ │ SimpleLiteLLM │ (Existing) │
│ │ Agent │ │
│ └────────┬────────┘ │
│ │ │
│ ┌────────▼────────┐ │
│ │ ADK Agent │ (New - No Tools) │
│ │ │ │
│ │ • LiteLLM │ │
│ │ • Streaming │ │
│ │ • Basic Q&A │ │
│ └────────┬────────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ LiteLLM │ │
│ └──────┬──────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ Ollama │ │
│ └─────────────┘ │
└─────────────────────────────────────────┘
```
### Next State (Phase 2)
```
┌─────────────────────────────────────────┐
│ Core-AI Service │
│ │
│ ┌────────────────┐ │
│ │ ADK Agent │ │
│ │ with Tools │ │
│ │ │ │
│ │ ┌──────────┐ │ │
│ │ │ Tools │──┼────► Core-API │
│ │ │ Registry │ │ (REST calls) │
│ │ └──────────┘ │ │
│ └────────┬───────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ LiteLLM │ │
│ └─────────────┘ │
└─────────────────────────────────────────┘
```
---
## Files Modified/Created
### Modified
- ✏️ `src/config.py` - Added ADK settings
- ✏️ `src/prompts.py` - Added ADK prompt
- ✏️ `main.py` - Updated import path
### Created
- 📄 `src/agents/__init__.py`
- 📄 `src/agents/simple.py` (moved from src/agent.py)
- 📄 `src/agents/adk_agent.py`
- 📄 `diagnostics/test_adk_direct.py`
- 📄 `tests/test_06_adk_setup.py`
- 📄 `ARCHITECTURE.md`
- 📄 `PHASE1_COMPLETE.md` (this file)
---
## Next Steps
### Rebuild & Test Phase 1
```bash
# Rebuild container
docker compose -f /mnt/media/Projects/portainer-core/stacks/core-ai.yml build
# Restart
docker compose -f /mnt/media/Projects/portainer-core/stacks/core-ai.yml up -d
# Test ADK
docker exec core-ai python diagnostics/test_adk_direct.py
# Run test suite
docker exec core-ai pytest tests/test_06_adk_setup.py -v -s
```
### Phase 2: Tool Integration
Once Phase 1 tests pass:
1. Create `src/tools/registry.py`
2. Implement REST-based tools
3. Create `tests/test_07_adk_tools.py`
4. Test tool registration and discovery
---
## Known Limitations (Phase 1)
⚠️ **No Tools Yet**
- ADK agent has no tools in Phase 1
- Can only do basic Q&A like SimpleLiteLLMAgent
- Tool calling framework is ready but unused
⚠️ **No HTTP Endpoints Yet**
- ADK agent not exposed via HTTP
- Only testable via diagnostics
- Phase 4 will add API routes
⚠️ **No Core-API Integration**
- Tools will call Core-API REST endpoints
- Integration happens in Phase 2
---
## Success Criteria for Phase 1
- [x] ADK imports successfully
- [x] ADKAgent class created
- [x] Agent initializes with Ollama/LiteLLM
- [x] Can answer simple questions
- [x] Streaming works
- [x] Non-streaming works
- [x] Diagnostic tool created
- [x] Test layer 6 created
- [ ] Tests pass in Docker container
**Status:** Implementation complete, awaiting rebuild and testing.
---
**Last Updated:** 2025-11-27
**Next Phase:** Tool Integration (Phase 2)
services/core-ai/PHASE1_SUCCESS.md
-258
View File
@@ -1,258 +0,0 @@
# Phase 1: ADK Agent Setup - SUCCESS ✅
**Date:** 2025-11-27
**Status:** COMPLETE AND WORKING
---
## 🎉 Achievement
**Core-AI now has TWO functional AI agents:**
1. ✅ **SimpleLiteLLMAgent** - Direct LiteLLM → Ollama (existing)
2. ✅ **ADKAgent** - Google ADK → LiteLLM → Ollama (NEW!)
Both agents successfully:
- Initialize properly
- Connect to Ollama
- Generate responses to queries
- Support streaming and non-streaming modes
---
## Test Results
### SimpleLiteLLMAgent (Existing - Still Working)
```
Query: "What is the capital of France?"
Response: "The capital of France is Paris."
Status: ✅ PASS
```
### ADKAgent (New - Now Working!)
```
Query: "What is the capital of France?"
Response: "I do not have access to real-time information..."
Status: ✅ WORKING (response quality can be improved)
Technical Status:
✅ Session creation
✅ Agent initialization
✅ Runner execution
✅ Event processing
✅ Response retrieval
```
---
## What Was Fixed
### Issue #1: Wrong Import Paths
**Problem:** Used non-existent `google.adk.llms.LiteLLM`
**Fix:** Changed to official API: `google.adk.models.lite_llm.LiteLlm`
### Issue #2: Wrong Execution Method
**Problem:** Tried to call `agent.run()` which doesn't exist
**Fix:** Used official pattern: `Runner.run_async()` with events
### Issue #3: Missing Session Management
**Problem:** ADK requires sessions but we didn't create them
**Fix:** Always create session before running agent
### Issue #4: Async/Await Issues
**Problem:** Forgot to `await` async session methods
**Fix:** Added `await` to all async calls
---
## Final Architecture (Phase 1)
```
┌─────────────────────────────────────────────────────────┐
│ Core-AI Service │
│ │
│ ┌──────────────────────┐ ┌───────────────────────┐ │
│ │ SimpleLiteLLMAgent │ │ ADKAgent │ │
│ │ (Simple Mode) │ │ (ADK Mode) │ │
│ │ │ │ │ │
│ │ • Direct LiteLLM │ │ • ADK Runtime │ │
│ │ • No tools │ │ • Runner + Sessions │ │
│ │ • Fast & lean │ │ • No tools (yet) │ │
│ └──────────┬───────────┘ └───────────┬───────────┘ │
│ │ │ │
│ └──────────┬───────────────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ LiteLLM │ │
│ └──────┬──────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ Ollama │ │
│ └──────┬──────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │Model (Gemma2)│ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
```
---
## Code Quality Improvements
### Documentation
- ✅ Official ADK documentation references in code
- ✅ Clear docstrings explaining parameters and returns
- ✅ Logging at all critical steps
### Error Handling
- ✅ Try/catch blocks around ADK operations
- ✅ Graceful fallbacks when no response
- ✅ Detailed error logging with stack traces
### Structure
- ✅ Agents separated into `src/agents/` directory
- ✅ Simple and ADK agents isolated from each other
- ✅ Clean imports with availability checks
---
## Files Created/Modified
### New Files
- 📄 `src/agents/__init__.py` - Agent exports
- 📄 `src/agents/simple.py` - SimpleLiteLLMAgent (moved)
- 📄 `src/agents/adk_agent.py` - ADKAgent (new)
- 📄 `diagnostics/test_adk_direct.py` - ADK diagnostic tool
- 📄 `tests/test_06_adk_setup.py` - ADK test layer
- 📄 `ARCHITECTURE.md` - Dual-mode architecture docs
- 📄 `PHASE1_COMPLETE.md` - Initial completion doc
- 📄 `PHASE1_SUCCESS.md` - This file
### Modified Files
- ✏️ `src/config.py` - Added ADK settings
- ✏️ `src/prompts.py` - Added ADK prompt variant
- ✏️ `main.py` - Updated imports
---
## Known Limitations (Phase 1)
### Response Quality
The ADK agent's responses are sometimes overly cautious:
- Says "I don't have access to real-time information" for basic facts
- Could be improved with better system prompts
- Model choice (Gemma2) may need tuning for better knowledge recall
**This is a prompt engineering issue, not a technical issue.**
### No Tools Yet
- ADK agent has framework for tools but none registered
- Phase 2 will add REST-based tools
- Tool calling capability exists but untested
### No HTTP Endpoints Yet
- ADK agent only accessible via Python imports
- Phase 4 will add `/v1/chat/adk` endpoint
- Currently only testable via diagnostics
---
## Next Steps
### Immediate (Optional Improvement)
- [ ] Improve ADK system prompt for better responses
- [ ] Test with different models (mistral, etc.)
- [ ] Add more test cases to test_06
### Phase 2: Tool Integration
- [ ] Create `src/tools/registry.py`
- [ ] Implement REST-based tools (call core-api)
- [ ] Register tools with ADK agent
- [ ] Create `tests/test_07_adk_tools.py`
### Phase 3: ADK Agent with Tools
- [ ] Test tool calling with simple tools
- [ ] Test multi-tool workflows
- [ ] Create `tests/test_08_adk_agent.py` and `test_09_tool_calling.py`
### Phase 4: API Routes
- [ ] Add `/v1/chat/simple` endpoint
- [ ] Add `/v1/chat/adk` endpoint
- [ ] Maintain `/v1/chat/completions` as alias
- [ ] Create `tests/test_10_adk_api.py`
---
## How to Test
### Quick Test
```bash
docker exec core-ai python -c "
import asyncio
from src.agents import ADKAgent
async def test():
agent = ADKAgent(tools=[])
response = await agent.chat_completion(
messages=[{'role': 'user', 'content': 'What is 2+2?'}]
)
print(f'Response: {response}')
asyncio.run(test())
"
```
### Full Diagnostic
```bash
docker exec core-ai python diagnostics/test_adk_direct.py
```
### Test Suite
```bash
docker exec core-ai pytest tests/test_06_adk_setup.py -v -s
```
---
## Lessons Learned
1. **Always check official docs** - Core-API implementation was broken, official docs were correct
2. **ADK requires specific patterns** - Runner + Sessions + Events, not just agent.run()
3. **Async/await matters** - Forgetting `await` causes silent failures
4. **Session management is mandatory** - ADK won't work without valid sessions
5. **Response quality ≠ technical success** - Integration works even if responses need tuning
---
## Success Criteria Met
- [x] ADK imports successfully
- [x] ADKAgent class created and working
- [x] Agent initializes with Ollama/LiteLLM
- [x] Can process queries and return responses
- [x] Streaming mode works (simulated)
- [x] Non-streaming mode works
- [x] Session management works
- [x] Runner execution works
- [x] Event processing works
- [x] Diagnostic tool created
- [x] Test layer 6 created
- [x] All tests can run (response quality separate)
**Phase 1 Status:** ✅ **COMPLETE AND FUNCTIONAL**
---
## Resources Used
- [Google ADK Python Docs](https://google.github.io/adk-docs/get-started/python/)
- [LiteLLM + ADK Tutorial](https://docs.litellm.ai/docs/tutorials/google_adk)
- [Building Local AI Agent with ADK](https://medium.com/@viplav.fauzdar/building-a-local-ai-agent-with-google-adk-litellm-and-ollama-6e907e2db268)
- [Ollama-Powered AI Agents](https://medium.com/@jageenshukla/how-to-build-ollama-powered-ai-agents-with-adk-tool-calling-and-mcp-integration-c25d98fc4816)
---
**Last Updated:** 2025-11-27
**Next Phase:** Tool Integration (Phase 2)
**Recommendation:** Proceed to Phase 2 or improve prompts for better response quality
services/core-ai/PHASE2_COMPLETE.md
-337
View File
@@ -1,337 +0,0 @@
# Phase 2: Tool Integration - COMPLETE ✓
**Date:** 2025-11-27
**Status:** COMPLETE AND TESTED
---
## 🎉 Achievement
**Core-AI now has a complete tool system:**
1. ✅ **Local Tools** - Time, date, and calculator utilities
2. ✅ **Tool Registry** - Central management system for all tools
3. ✅ **Swagger Discovery** - Dynamic tool creation from OpenAPI specs
4. ✅ **ADK Integration** - Tools work seamlessly with ADK agent
---
## Architecture
### Tool System Design
```
┌─────────────────────────────────────────────────────────────────┐
│ Core-AI Tool System │
│ │
│ ┌──────────────────┐ ┌──────────────────────────┐ │
│ │ Local Tools │ │ REST Tool Discovery │ │
│ │ │ │ │ │
│ │ • get_current_ │ │ • Fetch OpenAPI spec │ │
│ │ time() │ │ • Parse endpoints │ │
│ │ • get_current_ │ │ • Create dynamic tools │ │
│ │ date() │ │ • Register with ADK │ │
│ │ • calculate() │ │ │ │
│ │ • date ops │ │ Source: core-api │ │
│ └────────┬─────────┘ └────────────┬─────────────┘ │
│ │ │ │
│ └──────────────┬───────────────────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ Tool │ │
│ │ Registry │ │
│ └──────┬──────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ ADK Agent │ │
│ │ with Tools │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
### Tool Flow
1. **Registration Phase:**
- Local tools register via `@register_tool` decorator
- REST tools discovered from core-api's OpenAPI spec
- All tools added to central registry
2. **Conversion Phase:**
- Registry converts Python functions to ADK `FunctionTool` objects
- Type annotations mapped to ADK schema
- Descriptions extracted from docstrings
3. **Execution Phase:**
- ADK agent receives tool list during initialization
- Agent can call tools to answer user queries
- Tool calls logged and results returned to agent
---
## Test Results
### Layer 7: Tool Integration Tests
```bash
docker exec core-ai pytest tests/test_07_adk_tools.py -v
```
**Results:** ✅ **7/7 PASSED**
| Test | Status | Description |
|------|--------|-------------|
| `test_local_tools_registered` | ✅ PASS | All 5 local tools registered |
| `test_local_tool_execution` | ✅ PASS | Tools execute correctly |
| `test_calculator_security` | ✅ PASS | Calculator blocks dangerous expressions |
| `test_adk_tool_conversion` | ✅ PASS | Tools convert to ADK format |
| `test_adk_agent_with_tools` | ✅ PASS | Agent initializes with tools |
| `test_tool_calling_integration` | ✅ PASS | Agent uses tools to answer queries |
| `test_date_tools` | ✅ PASS | Date manipulation tools work |
---
## What Was Implemented
### 1. Tool Registry System ✓
**File:** `src/tools/registry.py`
**Features:**
- Tool registration via `@register_tool` decorator
- Conversion to ADK `FunctionTool` format
- Logging decorator for all tool calls
- Dynamic REST tool creation from OpenAPI specs
**Key Functions:**
```python
@register_tool
async def my_tool(param: str) -> str:
"""Tool description"""
return result
# Get all registered tools
tools = get_all_tools()
# Get ADK-compatible tools
adk_tools = get_agent_tools()
# Discover tools from core-api
await discover_and_register_tools(base_url)
```
### 2. Local Tools ✓
**File:** `src/tools/local.py`
**Tools Implemented:**
- `get_current_time()` - Get current UTC time
- `get_current_date()` - Get current date
- `calculate(expression: str)` - Safe math calculator
- `add_days_to_date(date: str, days: int)` - Date arithmetic
- `calculate_date_difference(date1: str, date2: str)` - Date comparison
**Security:**
- Calculator blocks dangerous operations (`exec`, `eval`, `import`, etc.)
- Restricted eval namespace (no builtins)
- Input validation for all tools
### 3. Swagger/OpenAPI Discovery ✓
**File:** `src/tools/registry.py` (functions: `fetch_openapi_spec`, `create_rest_tool`, `discover_and_register_tools`)
**Features:**
- Fetch OpenAPI spec from multiple possible endpoints
- Parse paths and operations
- Extract parameters (path, query, body)
- Generate async functions that call REST endpoints
- Register dynamically created tools
**Usage:**
```python
# Discover and register all tools from core-api
tool_count = await discover_and_register_tools("http://core-api:8083")
print(f"Registered {tool_count} REST tools")
```
### 4. ADK Agent Integration ✓
**File:** `src/agents/adk_agent.py`
**Updates:**
- Added `discover_tools` parameter to `__init__`
- Automatic tool loading from registry
- Tools passed to ADK Agent constructor
**Usage:**
```python
# Agent with local tools only
agent = ADKAgent(discover_tools=True)
# Agent without tools
agent = ADKAgent(discover_tools=False)
# Agent with explicit tools
agent = ADKAgent(tools=[my_tool])
```
### 5. Test Suite ✓
**Files Created:**
- `tests/test_07_adk_tools.py` - Unit tests for tool system
- `diagnostics/test_adk_tools.py` - Diagnostic tool for testing
---
## Files Created/Modified
### New Files
- 📄 `src/tools/__init__.py` - Tool module exports
- 📄 `src/tools/registry.py` - Tool registration and discovery
- 📄 `src/tools/local.py` - Local utility tools
- 📄 `tests/test_07_adk_tools.py` - Tool layer tests
- 📄 `diagnostics/test_adk_tools.py` - Tool diagnostic
- 📄 `PHASE2_COMPLETE.md` - This file
### Modified Files
- ✏️ `src/agents/adk_agent.py` - Added tool discovery support
- ✏️ `stacks/core-ai.yml` - Removed obsolete `version` field
---
## Key Technical Decisions
### 1. Tool Architecture
**Decision:** Local tools in core-ai, REST tools in core-api
**Rationale:**
- Local tools (time, calc) don't need network calls
- REST tools from core-api enable separation of concerns
- Dynamic discovery means core-api can add tools without core-ai changes
### 2. Registry Pattern
**Decision:** Central registry with decorator-based registration
**Rationale:**
- Simple developer experience (`@register_tool`)
- Automatic discovery at import time
- Single source of truth for all tools
### 3. Security Model
**Decision:** Restricted eval for calculator, no default parameters
**Rationale:**
- ADK doesn't support default parameter values
- Calculator must block dangerous operations
- Whitelist approach for allowed functions
### 4. OpenAPI Discovery
**Decision:** Dynamic tool generation from Swagger docs
**Rationale:**
- Self-documenting API becomes self-registering tools
- No code duplication between API and tools
- Automatic parameter type mapping
---
## Known Limitations (Phase 2)
### No HTTP Endpoints Yet
- Tools only accessible via Python imports
- Phase 4 will add `/v1/chat/adk` endpoint
- Currently testable via diagnostics only
### Core-API Discovery Not Tested
- REST tool discovery implemented but not tested with real core-api
- Needs core-api to have OpenAPI documentation
- Will test in Phase 3 when core-api is ready
### Limited Tool Coverage
- Only 5 local tools implemented
- More tools can be added as needed
- REST tools depend on core-api implementation
---
## Testing
### Run All Tests
```bash
# Layer 7: Tool integration
docker exec core-ai pytest tests/test_07_adk_tools.py -v -s
# Full diagnostic
docker exec core-ai python diagnostics/test_adk_tools.py
# Quick test
docker exec core-ai python -c "
from src.tools.local import calculate
import asyncio
result = asyncio.run(calculate('2 + 2'))
print(f'Result: {result}')
"
```
### Example: Test Tool with Agent
```python
from src.agents import ADKAgent
import asyncio
async def test():
agent = ADKAgent(discover_tools=True)
response = await agent.chat_completion(
messages=[{"role": "user", "content": "What is 15 + 27? Use the calculator."}]
)
print(response)
asyncio.run(test())
```
---
## Next Steps
### Phase 3: Core-API Integration
1. Add OpenAPI documentation to core-api
2. Test REST tool discovery from core-ai
3. Verify tool calling works across services
4. Add authentication/authorization for tool endpoints
### Phase 4: API Routes
1. Add `/v1/chat/simple` endpoint (SimpleLiteLLMAgent)
2. Add `/v1/chat/adk` endpoint (ADKAgent with tools)
3. Keep `/v1/chat/completions` as default alias
4. Create test_10_adk_api.py for HTTP testing
### Optional Improvements
- Add more local tools (system info, file ops)
- Implement tool result caching
- Add tool execution timeout limits
- Implement tool permission system
---
## Success Criteria Met
- [x] Tool registry system created
- [x] Local tools implemented (time, date, calculator)
- [x] Tools registered automatically via decorator
- [x] ADK tool conversion working
- [x] Agent can use tools
- [x] OpenAPI/Swagger discovery implemented
- [x] Security measures in place (calculator safety)
- [x] Test suite created and passing (7/7)
- [x] Diagnostic tool created
- [x] Documentation complete
**Phase 2 Status:** ✅ **COMPLETE AND FULLY TESTED**
---
## Resources
- [Google ADK Tool Documentation](https://google.github.io/adk-docs/python/tools/)
- [OpenAPI Specification](https://swagger.io/specification/)
- [FastAPI OpenAPI Support](https://fastapi.tiangolo.com/how-to/extending-openapi/)
---
**Last Updated:** 2025-11-27
**Next Phase:** Core-API Integration (Phase 3) or API Routes (Phase 4)
**Recommendation:** Add OpenAPI docs to core-api, then test full tool integration
services/core-ai/PHASE4_COMPLETE.md
-424
View File
@@ -1,424 +0,0 @@
# Phase 4: API Routes - COMPLETE ✓
**Date:** 2025-11-27
**Status:** COMPLETE WITH KNOWN ISSUES
---
## 🎉 Achievement
**Core-AI now has complete HTTP API endpoints:**
1. ✅ `/v1/chat/completions` - Default endpoint (simple agent)
2. ✅ `/v1/chat/simple` - Explicit simple agent (no tools)
3. ✅ `/v1/chat/adk` - ADK agent with tools
4. ✅ `/v1/tools` - List all available tools
5. ✅ `/health` - Enhanced health check with agent status
**Test Results:** 7/8 tests passing (87.5% pass rate)
---
## API Endpoints
### POST /v1/chat/completions
**Description:** Default chat endpoint (uses SimpleLiteLLMAgent)
**Status:** ✅ Working
**Request:**
```json
{
"messages": [
{"role": "user", "content": "What is 2+2?"}
],
"stream": false
}
```
**Response:**
```json
{
"id": "chatcmpl-abc123",
"object": "chat.completion",
"created": 1701234567,
"model": "default_model",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": "4"},
"finish_reason": "stop"
}]
}
```
### POST /v1/chat/simple
**Description:** Explicit simple agent endpoint (no tools)
**Status:** ✅ Working
**Request:**
```json
{
"messages": [{"role": "user", "content": "Hello"}],
"stream": false
}
```
**Response:** Same format as `/v1/chat/completions` with `"model": "simple"`
### POST /v1/chat/adk
**Description:** ADK agent endpoint with tool support
**Status:** ⚠️ Working but tool execution needs improvement
**Request:**
```json
{
"messages": [{"role": "user", "content": "What is the current date?"}],
"stream": false,
"enable_tools": true
}
```
**Response:**
```json
{
"id": "chatcmpl-xyz789",
"object": "chat.completion",
"created": 1701234567,
"model": "adk",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": "..."},
"finish_reason": "stop"
}],
"tools_enabled": true,
"tools_count": 5
}
```
**Parameters:**
- `enable_tools` (boolean, default: true) - Enable/disable tool usage
- `stream` (boolean, default: false) - Enable streaming responses
### GET /v1/tools
**Description:** List all available tools
**Status:** ✅ Working
**Response:**
```json
{
"tools": [
{
"name": "get_current_time",
"description": "Get the current time in UTC timezone...",
"type": "local"
},
...
],
"count": 5,
"adk_available": true
}
```
### GET /health
**Description:** Enhanced health check
**Status:** ✅ Working
**Response:**
```json
{
"status": "ok",
"service": "core-ai",
"agents": {
"simple": true,
"adk": true
},
"tools_count": 5
}
```
---
## Test Results
### Layer 10: API Integration Tests
```bash
docker exec core-ai pytest tests/test_10_adk_api.py -v
```
**Results:** ✅ **7/8 PASSED** (87.5%)
| Test | Status | Description |
|------|--------|-------------|
| `test_health_check` | ✅ PASS | Health endpoint returns correct status |
| `test_list_tools` | ✅ PASS | Tools listing endpoint works |
| `test_chat_completions_simple` | ✅ PASS | Default endpoint works |
| `test_chat_simple_endpoint` | ✅ PASS | Simple agent endpoint works |
| `test_chat_adk_endpoint` | ❌ FAIL | ADK endpoint timeout (30s) |
| `test_chat_adk_with_calculator` | ✅ PASS | ADK with calculator works |
| `test_streaming_simple` | ✅ PASS | Streaming responses work |
| `test_adk_without_tools` | ✅ PASS | ADK without tools works |
---
## What Was Implemented
### 1. HTTP Endpoints ✓
**File:** `main.py`
**New Handlers:**
- `chat_simple()` - SimpleLiteLLMAgent endpoint
- `chat_adk()` - ADKAgent endpoint with tool support
- `list_tools()` - Tool listing endpoint
- Enhanced `health_check()` - Shows agent and tool status
**Features:**
- OpenAI-compatible response format
- Streaming and non-streaming support
- Tool enable/disable control
- Proper error handling and logging
- Request logging with agent identifiers
### 2. Test Suite ✓
**File:** `tests/test_10_adk_api.py`
**Tests Created:**
- Health check validation
- Tools listing validation
- Simple agent endpoint testing
- ADK agent endpoint testing
- Streaming response testing
- Tool execution testing
- Error handling testing
---
## Architecture
### Request Flow
```
┌─────────────────────────────────────────────────────────────┐
│ HTTP Client │
└────────────┬────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ aiohttp Server │
│ (main.py) │
│ │
│ ┌──────────────────┐ ┌──────────────────────────────┐ │
│ │ /v1/chat/ │ │ /v1/chat/adk │ │
│ │ completions │ │ │ │
│ │ /v1/chat/simple │ │ • enable_tools param │ │
│ │ │ │ • Tool discovery │ │
│ │ → Simple Agent │ │ → ADK Agent │ │
│ └──────────────────┘ └──────────────────────────────┘ │
│ │
│ ┌──────────────────┐ ┌──────────────────────────────┐ │
│ │ /v1/tools │ │ /health │ │
│ │ → List tools │ │ → Status check │ │
│ └──────────────────┘ └──────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
```
### Endpoint Comparison
| Feature | /v1/chat/simple | /v1/chat/adk |
|---------|----------------|--------------|
| Agent | SimpleLiteLLMAgent | ADKAgent |
| Tools | ❌ No | ✅ Yes (optional) |
| Performance | Fast | Slower (with tools) |
| Streaming | ✅ Yes | ✅ Yes |
| Use Case | Quick Q&A | Complex tasks with tools |
---
## Usage Examples
### Simple Query (No Tools)
```bash
curl -X POST http://localhost:8086/v1/chat/simple \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": "What is 2+2?"}],
"stream": false
}'
```
### ADK Query (With Tools)
```bash
curl -X POST http://localhost:8086/v1/chat/adk \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": "What is the current date?"}],
"stream": false,
"enable_tools": true
}'
```
### List Available Tools
```bash
curl -X GET http://localhost:8086/v1/tools
```
### Streaming Request
```bash
curl -N -X POST http://localhost:8086/v1/chat/simple \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": "Count to 5"}],
"stream": true
}'
```
---
## Known Issues & Limitations
### 1. ADK Tool Execution Timeout
**Issue:** `test_chat_adk_endpoint` times out after 30 seconds
**Impact:** Medium - ADK agent with tool discovery takes too long for some queries
**Symptoms:**
- Request times out waiting for response
- Happens when ADK tries to determine which tool to use
- Works fine when tools are disabled
**Possible Causes:**
- ADK runner processing all events before returning final response
- Tool call event handling incomplete
- Model taking too long to decide on tool usage
**Workaround:**
- Increase timeout to 60 seconds
- Disable tools for simple queries
- Use `/v1/chat/simple` for basic Q&A
**TODO:** Investigate ADK event loop and tool execution flow
### 2. Tool Call Format
**Issue:** ADK sometimes returns tool call JSON instead of executing tools
**Impact:** Low - Appears to be intermittent
**Symptoms:**
```json
{
"toolCalls": [{
"id": "call_xxx",
"type": "function",
"function": {"name": "get_current_date", "arguments": {}}
}]
}
```
**Possible Causes:**
- ADK runner not processing all events
- Breaking out of event loop too early
- Missing event type handling
**TODO:** Review adk_agent.py event processing logic
### 3. No REST Tool Discovery Yet
**Status:** Not implemented in this phase
**Impact:** Low - Phase 2 implemented the framework, Phase 3 will test it
**Next Steps:** Test with real core-api OpenAPI documentation
---
## Files Modified/Created
### Modified Files
- ✏️ `main.py` - Added 4 new endpoints and enhanced health check
### New Files
- 📄 `tests/test_10_adk_api.py` - API integration tests
- 📄 `PHASE4_COMPLETE.md` - This file
---
## Performance Metrics
### Response Times (Approximate)
- `/health`: < 50ms
- `/v1/tools`: < 100ms
- `/v1/chat/simple`: 1-5 seconds (depends on model)
- `/v1/chat/adk` (no tools): 2-8 seconds
- `/v1/chat/adk` (with tools): 5-30+ seconds
### Concurrent Requests
- Simple endpoint: Handles multiple concurrent requests well
- ADK endpoint: One request at a time recommended (caching helps)
---
## Next Steps
### Immediate Fixes
- [ ] Investigate and fix ADK tool execution timeout
- [ ] Improve ADK event processing to handle tool calls properly
- [ ] Add request timeout configuration
### Phase 5 (Future)
- [ ] Add authentication/authorization
- [ ] Add rate limiting
- [ ] Add request/response logging to database
- [ ] Add metrics/monitoring endpoints
- [ ] Implement conversation history persistence
### Phase 3 (Revisit)
- [ ] Test REST tool discovery with real core-api
- [ ] Add core-api OpenAPI documentation
- [ ] Verify cross-service tool calling
---
## Success Criteria
- [x] `/v1/chat/completions` endpoint working
- [x] `/v1/chat/simple` endpoint working
- [x] `/v1/chat/adk` endpoint working (with known issues)
- [x] `/v1/tools` endpoint working
- [x] Enhanced `/health` endpoint
- [x] Streaming support for all chat endpoints
- [x] OpenAI-compatible response format
- [x] Test suite created (8 tests)
- [x] 87.5% test pass rate (7/8 passing)
- [ ] 100% test pass rate (pending timeout fix)
**Phase 4 Status:** ✅ **COMPLETE WITH KNOWN ISSUES**
---
## Testing
### Quick Manual Tests
```bash
# Health check
curl -s http://localhost:8086/health | jq .
# List tools
curl -s http://localhost:8086/v1/tools | jq .tools[].name
# Simple chat
curl -s -X POST http://localhost:8086/v1/chat/simple \
-H "Content-Type: application/json" \
-d '{"messages": [{"role": "user", "content": "Hello"}], "stream": false}' \
| jq .choices[0].message.content
# ADK chat (no tools)
curl -s -X POST http://localhost:8086/v1/chat/adk \
-H "Content-Type: application/json" \
-d '{"messages": [{"role": "user", "content": "Hello"}], "stream": false, "enable_tools": false}' \
| jq .choices[0].message.content
```
### Full Test Suite
```bash
docker exec core-ai pytest tests/test_10_adk_api.py -v -s
```
---
**Last Updated:** 2025-11-27
**Next Phase:** Fix tool execution issues, then proceed to Phase 3 (Core-API integration)
**Recommendation:** Address ADK timeout issue before production use
services/core-ai/README.md
+430 -208
View File
@@ -1,23 +1,241 @@
# Core-AI Service
Simplified AI service for testing LiteLLM → Ollama → Model integration without ADK complexity.
## Purpose
This service strips away the ADK layer to isolate and debug the fundamental LiteLLM/Ollama integration. It provides:
- **Direct LiteLLM integration** - No ADK overhead
- **OpenAI-compatible API** - Drop-in replacement for testing
- **Comprehensive diagnostics** - Layered testing to identify issues
- **Minimal complexity** - Easy to understand and debug
AI agent service built on PydanticAI for infrastructure management and automation.
## Architecture
Core-AI provides two agents with distinct capabilities:
```
HTTP Request → SimpleLiteLLMAgent → LiteLLM → Ollama → Model → Response
┌─────────────────────────────────────────────────────────────┐
│ Core-AI Service │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────┐ ┌──────────────────────┐ │
│ │ PydanticAgent │ │ SimpleLiteLLMAgent │ │
│ │ (Primary) │ │ (Fallback) │ │
│ │ │ │ │ │
│ │ • Tool calling │ │ • No tools │ │
│ │ • Memory (3-tier)│ │ • Direct LiteLLM │ │
│ │ • OpenAPI tools │ │ • Minimal overhead │ │
│ └────────┬─────────┘ └──────────┬───────────┘ │
│ │ │ │
│ └──────────┬───────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────┐ │
│ │ PydanticAI Runtime │ │
│ │ (Ollama backend) │ │
│ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
```
**Bypassed:** Google ADK, tool calling, complex orchestration
### PydanticAgent (Primary)
**Endpoint:** `/v1/chat/completions` (default)
Advanced agent using the PydanticAI framework with:
- **Tool Calling:** Automatic function calling with proper validation
- **Memory System:** 3-tier conversation memory (buffer + Qdrant)
- **Local Tools:** Time, calculations, web search (SearXNG)
- **OpenAPI Tools:** Auto-discovered from core-api infrastructure endpoints
- **Streaming Support:** Server-sent events for real-time responses
### SimpleLiteLLMAgent (Fallback)
**Endpoint:** `/v1/chat/simple`
Lightweight agent for direct LLM interaction:
- **No Tools:** Pure conversational mode
- **Direct LiteLLM:** Minimal abstraction layer
- **No Memory:** Stateless request/response
- **Low Latency:** Fastest response times
## Tool System
### Local Tools
Built-in utilities available immediately (defined in `src/tools/local.py`):
- `get_current_time(timezone)` - Timezone-aware time with IANA timezone support
- `get_current_date()` - Current date in ISO format
- `calculate(expression)` - Safe mathematical calculations
- `calculate_date_difference(date1, date2)` - Date arithmetic
- `add_days_to_date(date, days)` - Date manipulation
- `web_search(query, category, max_results)` - SearXNG metasearch integration
### OpenAPI Discovery
Dynamically discovers infrastructure tools from core-api's OpenAPI spec:
- **Auto-Discovery:** Fetches `/openapi.json` on startup
- **REST Mapping:** Converts endpoints to callable functions
- **Prefixed Names:** Tools prefixed with service name (e.g., `core-api__list_containers`)
- **Type Safety:** Preserves parameter types and validation
**Configuration:**
```bash
OPENAPI_ENABLED=true
OPENAPI_ENDPOINTS=http://core-api:8083/openapi.json
```
**List Available Tools:**
```bash
curl http://localhost:8086/v1/tools
```
## Memory System
3-tier multi-tenant memory with per-user data isolation:
### Tier 1: Conversation Buffer (RAM)
- **Storage:** In-memory per-user buffers
- **Scope:** Recent N turns (configurable, default: 10)
- **Speed:** Instant access
- **Purpose:** Fast context for ongoing conversations
### Tier 2: Persistent Storage (Qdrant)
- **Storage:** Per-user Qdrant collections
- **Scope:** Complete conversation history
- **Speed:** Fast retrieval by conversation ID
- **Purpose:** Conversation continuity across sessions
### Tier 3: Semantic Search (Qdrant)
- **Storage:** Same as Tier 2 with vector embeddings
- **Scope:** Cross-conversation semantic search
- **Speed:** Sub-second similarity search
- **Purpose:** Contextual recall across all user conversations
### Multi-Tenancy
- **Per-User Collections:** Each user gets isolated Qdrant collection
- **User ID Format:** Sanitized email (`username_at_domain_com`)
- **GDPR Compliance:** Complete user data deletion support
- **Automatic Isolation:** No cross-user data leakage
**Memory Configuration:**
```bash
MEMORY_ENABLED=true
MEMORY_TIER1_SIZE=10
QDRANT_URL=http://qdrant:6333
EMBEDDING_MODEL=nomic-embed-text
DEFAULT_USER_ID=llmdefault_at_schweitz_net
```
## API Endpoints
### Chat Completions
**POST /v1/chat/completions** (Default: PydanticAI)
OpenAI-compatible chat endpoint using PydanticAgent.
**Request:**
```json
{
"messages": [
{"role": "user", "content": "What containers are running?"}
],
"conversation_id": "optional-conversation-id",
"enable_tools": true,
"stream": false
}
```
**Response:**
```json
{
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "I found 5 running containers..."
},
"finish_reason": "stop"
}],
"model": "pydantic",
"tools_enabled": true,
"tools_count": 12
}
```
**Streaming:** Set `"stream": true` for SSE response
### Simple Chat
**POST /v1/chat/simple**
No-tools fallback endpoint using SimpleLiteLLMAgent.
Same request/response format as above, but `tools_enabled` will be `false`.
### List Models
**GET /v1/models**
Returns available agent types:
- `pydantic` - PydanticAgent (primary)
- `simple` - SimpleLiteLLMAgent (fallback)
### List Tools
**GET /v1/tools**
Returns all available tools (local + discovered OpenAPI tools).
### Health Check
**GET /health**
Service health status with agent availability.
## Configuration
All configuration via environment variables (see `src/config.py`):
### Core Settings
| Variable | Default | Description |
|----------|---------|-------------|
| `HOST` | `0.0.0.0` | Server host |
| `PORT` | `8086` | Server port |
| `LOG_LEVEL` | `INFO` | Logging level |
### Ollama Integration
| Variable | Default | Description |
|----------|---------|-------------|
| `OLLAMA_BASE_URL` | `http://ollama:11434` | Ollama API URL |
| `AGENT_MODEL` | `mistral-nemo:latest` | Primary model (tool-calling optimized) |
| `OLLAMA_TIMEOUT` | `300` | Request timeout (seconds) |
### System Prompts
| Variable | Default | Description |
|----------|---------|-------------|
| `SYSTEM_PROMPT_VARIANT` | `minimal_agent` | Prompt for SimpleLiteLLMAgent |
| `PYDANTIC_SYSTEM_PROMPT_VARIANT` | `pydantic_agent` | Prompt for PydanticAgent |
### Tool Discovery
| Variable | Default | Description |
|----------|---------|-------------|
| `OPENAPI_ENABLED` | `true` | Enable OpenAPI tool discovery |
| `OPENAPI_ENDPOINTS` | `http://core-api:8083/openapi.json` | OpenAPI spec URLs (comma-separated) |
### Memory System
| Variable | Default | Description |
|----------|---------|-------------|
| `MEMORY_ENABLED` | `true` | Enable conversation memory |
| `MEMORY_TIER1_SIZE` | `10` | Max turns in RAM buffer |
| `QDRANT_URL` | `http://qdrant:6333` | Qdrant vector DB URL |
| `QDRANT_COLLECTION_PREFIX` | `core_ai_user` | Prefix for user collections |
| `EMBEDDING_MODEL` | `nomic-embed-text` | Ollama embedding model |
| `EMBEDDING_DIMENSION` | `768` | Embedding vector size |
| `DEFAULT_USER_ID` | `llmdefault_at_schweitz_net` | Default user (until auth integration) |
## Quick Start
@@ -29,158 +247,83 @@ pip install -r requirements.txt
### 2. Configure Environment
Create `.env` file or set environment variables:
Create `.env` file:
```bash
OLLAMA_BASE_URL=http://ollama:11434
AGENT_MODEL=gemma2:9b-instruct-q5_K_M
SYSTEM_PROMPT_VARIANT=minimal_agent
HOST=0.0.0.0
PORT=8086
AGENT_MODEL=mistral-nemo:latest
QDRANT_URL=http://qdrant:6333
MEMORY_ENABLED=true
OPENAPI_ENABLED=true
```
### 3. Run Diagnostics
```bash
# Check Ollama connectivity
python diagnostics/check_ollama.py
# Test direct LiteLLM
python diagnostics/test_litellm_direct.py
# Run full test suite
bash tests/run_all_tests.sh
```
### 4. Start Service
### 3. Start Service
```bash
python main.py
```
Service will be available at `http://localhost:8086`
Service available at `http://localhost:8086`
### 5. Test It
### 4. Test Chat
```bash
curl -X POST http://localhost:8086/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{
"messages": [
{"role": "user", "content": "What is the capital of France?"}
]
{"role": "user", "content": "What time is it in Amsterdam?"}
],
"enable_tools": true
}'
```
## API Endpoints
### `GET /health`
Health check endpoint
**Response:**
```json
{
"status": "ok",
"service": "core-ai"
}
```
### `POST /v1/chat/completions`
OpenAI-compatible chat completions endpoint
**Request:**
```json
{
"model": "test",
"messages": [
{"role": "user", "content": "Your question here"}
],
"stream": false
}
```
**Response (non-streaming):**
```json
{
"id": "chatcmpl-...",
"object": "chat.completion",
"created": 1234567890,
"model": "test",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "Response here"
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
}
```
**Streaming:** Set `"stream": true` for Server-Sent Events response
## Project Structure
```
services/core-ai/
├── main.py # HTTP server (aiohttp)
├── src/
│ ├── agent.py # SimpleLiteLLMAgent
│ ├── config.py # Configuration (Pydantic)
│ ├── prompts.py # System prompts
│ └── tools.py # (Unused in this version)
├── diagnostics/
│ ├── check_ollama.py # Ollama connectivity check
│ └── test_litellm_direct.py # Direct LiteLLM test
├── tests/
│ ├── test_01_environment.py # Config tests
│ ├── test_02_litellm_raw.py # Raw LiteLLM tests
│ ├── test_03_message_format.py # Message formatting
│ ├── test_04_agent.py # Agent logic tests
│ ├── test_05_api.py # API endpoint tests
│ ├── run_all_tests.sh # Run all tests
│ └── README.md # Test documentation
├── requirements.txt
├── Dockerfile
└── README.md (this file)
```
## Configuration
Configuration is managed via `src/config.py` using Pydantic Settings.
### Environment Variables
| Variable | Default | Description |
|----------|---------|-------------|
| `HOST` | `0.0.0.0` | Server host |
| `PORT` | `8086` | Server port |
| `OLLAMA_BASE_URL` | `http://ollama:11434` | Ollama API URL |
| `AGENT_MODEL` | `gemma2:9b-instruct-q5_K_M` | Model name |
| `SYSTEM_PROMPT_VARIANT` | `minimal_agent` | Prompt variant to use |
| `DEBUG` | `false` | Enable debug mode |
| `LOG_LEVEL` | `INFO` | Logging level |
The agent will automatically use the `get_current_time` tool.
## Testing
See [tests/README.md](tests/README.md) for comprehensive testing documentation.
### Unit Tests
**Quick test:**
```bash
bash tests/run_all_tests.sh
# Run all tests
pytest tests/ -v
# Run specific test suite
pytest tests/test_ai_flow_quality.py -v
# Run with coverage
pytest tests/ --cov=src --cov-report=html
```
This runs 5 layers of tests to isolate issues:
1. Environment & Configuration
2. Raw LiteLLM Connection
3. Message Formatting
4. Agent Logic
5. API Integration
### Integration Tests
Quality tests for end-to-end AI flows:
```bash
pytest tests/test_ai_flow_quality.py -v
```
See `tests/QUALITY_TESTS.md` for test documentation.
### Manual Testing
```bash
# Test PydanticAgent (with tools)
curl -X POST http://localhost:8086/v1/chat/completions \
-H 'Content-Type: application/json' \
-d '{"messages": [{"role": "user", "content": "Calculate 123 * 456"}]}'
# Test SimpleLiteLLMAgent (no tools)
curl -X POST http://localhost:8086/v1/chat/simple \
-H 'Content-Type: application/json' \
-d '{"messages": [{"role": "user", "content": "Hello!"}]}'
# List available tools
curl http://localhost:8086/v1/tools
# Health check
curl http://localhost:8086/health
```
## Docker Deployment
@@ -197,7 +340,8 @@ docker run -d \
--name core-ai \
-p 8086:8086 \
-e OLLAMA_BASE_URL=http://ollama:11434 \
-e AGENT_MODEL=gemma2:9b-instruct-q5_K_M \
-e QDRANT_URL=http://qdrant:6333 \
-e AGENT_MODEL=mistral-nemo:latest \
--network docker-dataplane \
core-ai:latest
```
@@ -208,106 +352,184 @@ docker run -d \
docker-compose -f ../../stacks/core-ai.yml up
```
## Troubleshooting
## Project Structure
### Service won't start
1. Check logs: `docker logs core-ai`
2. Verify Ollama is running: `docker ps | grep ollama`
3. Run diagnostics: `python diagnostics/check_ollama.py`
### No response or timeout
1. Check Ollama logs: `docker logs ollama`
2. Model may be loading (first run takes 30-60s)
3. Verify model exists: `docker exec ollama ollama list`
4. Test directly: `docker exec ollama ollama run gemma2:9b-instruct-q5_K_M "test"`
### Wrong or empty responses
1. Check system prompt is loaded (see agent logs)
2. Verify prompt variant exists in `src/prompts.py`
3. Run Layer 3 tests: `pytest tests/test_03_message_format.py -v`
### Connection refused
1. Check network: `docker network inspect docker-dataplane`
2. Verify both services are on the same network
3. Try using container IP instead of hostname
```
services/core-ai/
├── main.py # HTTP server (aiohttp)
├── src/
│ ├── agents/
│ │ ├── __init__.py # Agent exports
│ │ ├── pydantic_agent.py # PydanticAgent (primary)
│ │ └── simple.py # SimpleLiteLLMAgent (fallback)
│ ├── memory/
│ │ ├── manager.py # Multi-tenant memory manager
│ │ ├── tier1_buffer.py # RAM conversation buffer
│ │ ├── qdrant_memory.py # Qdrant persistent + semantic
│ │ ├── base.py # Base memory interfaces
│ │ └── schemas.py # Memory data schemas
│ ├── tools/
│ │ ├── local.py # Local utility tools
│ │ ├── openapi_discovery.py # OpenAPI tool discovery
│ │ └── registry.py # Tool registration system
│ ├── config.py # Configuration (Pydantic Settings)
│ ├── prompts.py # System prompts
│ └── utils.py # Utilities
├── tests/
│ ├── test_ai_flow_quality.py # End-to-end AI quality tests
│ └── QUALITY_TESTS.md # Test documentation
├── requirements.txt
├── Dockerfile
└── README.md (this file)
```
## Development
### Adding New Prompts
### Adding Local Tools
Edit `src/tools/local.py`:
```python
from src.tools.registry import register_tool
@register_tool
async def my_new_tool(param: str) -> str:
"""
Tool description for LLM.
Args:
param: Parameter description
Returns:
Result description
"""
# Implementation
return f"Result: {param}"
```
Tool automatically available to PydanticAgent.
### Adding OpenAPI Sources
Add endpoints to configuration:
```bash
OPENAPI_ENDPOINTS=http://core-api:8083/openapi.json,http://automation:8080/openapi.json
```
Tools auto-discovered on startup with service prefix:
- `core-api__list_containers`
- `automation__deploy_stack`
### Modifying System Prompts
Edit `src/prompts.py`:
```python
PROMPTS = {
"minimal_agent": "You are a helpful assistant.",
"my_new_prompt": "Your custom system prompt here."
"pydantic_agent": "Your custom PydanticAgent prompt...",
"minimal_agent": "Your custom SimpleLiteLLMAgent prompt..."
}
```
Update environment variable:
Update environment:
```bash
SYSTEM_PROMPT_VARIANT=my_new_prompt
PYDANTIC_SYSTEM_PROMPT_VARIANT=pydantic_agent
```
### Modifying Agent Behavior
### Memory System Usage
Edit `src/agent.py` - specifically the `SimpleLiteLLMAgent` class.
Memory automatically managed per user:
**Key methods:**
- `__init__()` - Initialization and configuration
- `chat()` - Streaming chat handler
- `chat_completion()` - Non-streaming completion handler
```python
from src.memory import get_memory_manager_for_user
### Adding Tests
# Get user's memory manager
memory = get_memory_manager_for_user(user_id="user_at_example_com")
Add to appropriate test layer in `tests/`:
- Configuration changes → `test_01_environment.py`
- LiteLLM behavior → `test_02_litellm_raw.py`
- Message formatting → `test_03_message_format.py`
- Agent logic → `test_04_agent.py`
- API changes → `test_05_api.py`
# Memory automatically used by PydanticAgent when conversation_id provided
# See: src/agents/pydantic_agent.py
```
## Comparison with Core-API
## Troubleshooting
| Feature | Core-AI | Core-API |
|---------|---------|----------|
| **ADK Integration** | ❌ No | ✅ Yes |
| **Tool Calling** | ❌ No | ✅ Yes |
| **System Orchestration** | ❌ No | ✅ Yes |
| **Complexity** | Low | High |
| **Purpose** | Debugging | Production |
| **Direct LiteLLM** | ✅ Yes | ❌ No |
| **Diagnostics** | ✅ Comprehensive | Limited |
### PydanticAI Not Available
## Next Steps
**Error:** `PydanticAI not available. Install with: pip install pydantic-ai`
### If Tests Pass
**Solution:**
```bash
pip install pydantic-ai
```
1. ✅ Foundation is solid
2. Consider migrating fixes to core-api
3. Add ADK layer back in phases
4. Test tool calling integration
### Tools Not Discovered
### If Tests Fail
**Issue:** `/v1/tools` returns empty list or only local tools
1. Run diagnostics to identify layer
2. Fix that specific layer
3. Re-run tests
4. Proceed once all pass
**Check:**
1. Verify `OPENAPI_ENABLED=true`
2. Check core-api is running: `curl http://core-api:8083/openapi.json`
3. Review logs for discovery errors: `docker logs core-ai`
### Memory Errors
**Issue:** Memory operations failing
**Check:**
1. Verify Qdrant running: `curl http://qdrant:6333/collections`
2. Check embedding model available: `docker exec ollama ollama list | grep nomic-embed-text`
3. Review logs for initialization errors
### Model Timeouts
**Issue:** Requests timing out
**Solutions:**
1. Increase timeout: `OLLAMA_TIMEOUT=600`
2. Use smaller model: `AGENT_MODEL=mistral-tools:7b`
3. Check GPU access: `docker exec ollama nvidia-smi`
### Tool Calling Failures
**Issue:** Agent not using tools correctly
**Check:**
1. Verify model supports tool calling: `mistral-nemo`, `mistral-tools:7b`
2. Test with `enable_tools=false` to isolate issue
3. Review tool logs: Look for `🔧 TOOL CALL:` in logs
## Model Recommendations
### For Tool Calling (PydanticAgent)
- **mistral-nemo:latest** (default) - Best balance
- **mistral-tools:7b** - Faster, less accurate
- **llama3.1:8b** - Good alternative
### For Simple Chat (SimpleLiteLLMAgent)
- **gemma2:9b** - Fast conversational
- **llama3.2:3b** - Minimal resources
- Any model works (no tool calling required)
## Migration Notes
This service has migrated from:
- **ADK (Agent Development Kit)** → PydanticAI
- **LangChain/LangGraph** → PydanticAI native
- **OllamaNativeAgent** → Removed (superseded by PydanticAgent)
All references to these frameworks have been removed. The codebase now exclusively uses PydanticAI for agent orchestration.
## Contributing
When making changes:
1. Run diagnostics first
2. Make changes
3. Run full test suite
4. Update relevant documentation
5. Test in Docker environment
1. Add tests in `tests/`
2. Update docstrings
3. Test with both agents (`/v1/chat/completions` and `/v1/chat/simple`)
4. Verify tool discovery works
5. Test memory persistence
## License
Part of the tower-of-joy project.
Part of the portainer-core project.
services/core-ai/diagnostics/test_litellm_direct.py
-144
View File
@@ -1,144 +0,0 @@
#!/usr/bin/env python3
"""
Diagnostic tool to test direct LiteLLM → Ollama communication.
This bypasses all abstractions and tests the raw integration.
Usage:
python diagnostics/test_litellm_direct.py
"""
import asyncio
import sys
import os
from pathlib import Path
# Add parent directory to path to import from src
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.config import get_settings
# Import LiteLLM
try:
import litellm
litellm.set_verbose = True
except ImportError:
print("✗ LiteLLM not installed. Run: pip install litellm")
sys.exit(1)
async def test_litellm_direct():
"""Test direct LiteLLM completion with Ollama"""
settings = get_settings()
print("=" * 70)
print("LITELLM DIRECT TEST")
print("=" * 70)
# Test configuration
model_name = settings.agent_model
litellm_model = f"ollama/{model_name}"
api_base = settings.ollama_base_url
print(f"\n1. Configuration")
print(f" LiteLLM Model: {litellm_model}")
print(f" API Base: {api_base}")
print(f" Temperature: 0.1")
# Test messages
test_cases = [
{
"name": "Simple question (no system prompt)",
"messages": [
{"role": "user", "content": "What is the capital of France? Answer in one word."}
]
},
{
"name": "Simple question (with system prompt)",
"messages": [
{"role": "system", "content": "You are a helpful assistant. Answer questions concisely."},
{"role": "user", "content": "What is the capital of France? Answer in one word."}
]
},
{
"name": "Math problem",
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is 2 + 2? Answer with just the number."}
]
}
]
# Run tests
for i, test_case in enumerate(test_cases, 1):
print(f"\n{'-' * 70}")
print(f"Test {i}/{len(test_cases)}: {test_case['name']}")
print(f"{'-' * 70}")
# Log messages being sent
print("\nMessages being sent:")
for j, msg in enumerate(test_case['messages']):
content_preview = msg['content'][:60] + "..." if len(msg['content']) > 60 else msg['content']
print(f" [{j}] {msg['role']}: {content_preview}")
try:
# Test non-streaming first
print("\n→ Testing non-streaming mode...")
response = await litellm.acompletion(
model=litellm_model,
messages=test_case['messages'],
api_base=api_base,
temperature=0.1,
stream=False
)
content = response.choices[0].message.content
finish_reason = response.choices[0].finish_reason
print(f"\n✓ Non-streaming response received:")
print(f" Content: {content}")
print(f" Finish reason: {finish_reason}")
print(f" Model: {response.model}")
# Test streaming
print("\n→ Testing streaming mode...")
stream_response = await litellm.acompletion(
model=litellm_model,
messages=test_case['messages'],
api_base=api_base,
temperature=0.1,
stream=True
)
chunks = []
chunk_count = 0
async for chunk in stream_response:
chunk_count += 1
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
full_content = "".join(chunks)
print(f"\n✓ Streaming response received:")
print(f" Content: {full_content}")
print(f" Chunks: {chunk_count}")
print(f"\n✓ Test {i} PASSED")
except Exception as e:
print(f"\n✗ Test {i} FAILED")
print(f" Error: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
return False
print("\n" + "=" * 70)
print("✓ ALL LITELLM TESTS PASSED!")
print("=" * 70)
print("\nNext steps:")
print(" 1. If this works, the LiteLLM → Ollama connection is solid")
print(" 2. Any issues are likely in the agent wrapper or API layer")
print(" 3. Run the full test suite: bash diagnostics/run_all_tests.sh")
return True
if __name__ == "__main__":
result = asyncio.run(test_litellm_direct())
sys.exit(0 if result else 1)
services/core-ai/main.py
+178 -331
View File
@@ -1,7 +1,7 @@
import os
import logging
import json
import time # Import time module
import time
from aiohttp import web
from aiohttp_cors import setup as cors_setup, ResourceOptions
from dotenv import load_dotenv
@@ -16,8 +16,6 @@ logger = logging.getLogger(__name__)
# Import the agent logic
from src.agents import (
get_simple_litellm_agent,
get_ollama_native_agent,
OLLAMA_NATIVE_AVAILABLE,
get_pydantic_agent,
PYDANTIC_AI_AVAILABLE
)
@@ -26,182 +24,8 @@ from src.utils import extract_user_id_from_request
async def chat_completions(request):
"""
Handles OpenAI-compatible chat completion requests using Ollama Native agent.
Default endpoint - uses Ollama Native Agent with tools enabled.
"""
if not OLLAMA_NATIVE_AVAILABLE:
return web.json_response({
"error": {"message": "Ollama Native agent not available"}
}, status=503)
try:
data = await request.json()
logger.info(f"[DEFAULT/OLLAMA_NATIVE] Received chat request")
# Extract relevant fields from the request
messages = data.get("messages")
model = data.get("model", "ollama-native")
stream = data.get("stream", False)
conversation_id = data.get("conversation_id")
enable_tools = data.get("enable_tools", True) # Tools enabled by default
if not messages:
raise web.HTTPBadRequest(reason="'messages' field is required")
# Get the agent instance (Ollama Native with working tool calling)
agent = get_ollama_native_agent(discover_tools=enable_tools)
# For non-streaming requests, collect the full response
if not stream:
response_content = await agent.chat_completion(
messages=messages,
conversation_id=conversation_id
)
return web.json_response({
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion",
"created": int(time.time()),
"model": "pydantic",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": response_content},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
},
"tools_enabled": enable_tools,
"tools_count": len(agent.tools_dict) if enable_tools else 0
})
else:
# Handle streaming response
response = web.StreamResponse(
status=200,
headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'}
)
await response.prepare(request)
async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True):
chunk_type = chunk.get("type", "content")
if chunk_type == "content":
json_chunk = {
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": "pydantic",
"choices": [{
"index": 0,
"delta": {"content": chunk.get("content", "")},
"finish_reason": chunk.get("finish_reason")
}]
}
await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode())
if chunk.get("finish_reason") == "stop":
break
elif chunk_type == "error":
error_chunk = {
"error": {"message": chunk.get("content", "Unknown error")}
}
await response.write(f"data: {json.dumps(error_chunk)}\n\n".encode())
break
await response.write(b"data: [DONE]\n\n")
await response.write_eof()
return response
except web.HTTPBadRequest as e:
logger.warning(f"Bad request: {e.reason}")
return web.json_response({"error": {"message": e.reason}}, status=400)
except Exception as e:
logger.exception("[DEFAULT/PYDANTIC_AI] Error during chat completion:")
return web.json_response({"error": {"message": str(e)}}, status=500)
async def chat_simple(request):
"""
Handles chat requests using SimpleLiteLLMAgent (no tools).
Endpoint: POST /v1/chat/simple
"""
try:
data = await request.json()
logger.info(f"[SIMPLE] Received chat request")
messages = data.get("messages")
model = data.get("model", "simple")
stream = data.get("stream", False)
conversation_id = data.get("conversation_id")
if not messages:
raise web.HTTPBadRequest(reason="'messages' field is required")
# Get SimpleLiteLLM agent
agent = get_simple_litellm_agent()
# Non-streaming response
if not stream:
response_content = await agent.chat_completion(
messages=messages,
conversation_id=conversation_id
)
return web.json_response({
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion",
"created": int(time.time()),
"model": "simple",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": response_content},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
})
else:
# Streaming response
response = web.StreamResponse(
status=200,
headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'}
)
await response.prepare(request)
async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True):
json_chunk = {
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": "simple",
"choices": [{
"index": 0,
"delta": {"content": chunk.get("content", "")},
"finish_reason": chunk.get("finish_reason")
}]
}
await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode())
if chunk.get("finish_reason") == "stop":
break
await response.write(b"data: [DONE]\n\n")
await response.write_eof()
return response
except web.HTTPBadRequest as e:
logger.warning(f"Bad request: {e.reason}")
return web.json_response({"error": {"message": e.reason}}, status=400)
except Exception as e:
logger.exception("[SIMPLE] Error during chat completion:")
return web.json_response({"error": {"message": str(e)}}, status=500)
async def chat_pydantic(request):
"""
Handles chat requests using PydanticAI Agent with tools.
Endpoint: POST /v1/chat/pydantic
Handles OpenAI-compatible chat completion requests using PydanticAI.
Default endpoint - uses PydanticAI with tools enabled.
"""
if not PYDANTIC_AI_AVAILABLE:
return web.json_response({
@@ -210,8 +34,9 @@ async def chat_pydantic(request):
try:
data = await request.json()
logger.info(f"[PYDANTIC_AI] Received chat request")
logger.info(f"[DEFAULT/PYDANTIC_AI] Received chat request")
# Extract relevant fields from the request
messages = data.get("messages")
model = data.get("model", "pydantic")
stream = data.get("stream", False)
@@ -224,25 +49,25 @@ async def chat_pydantic(request):
if not messages:
raise web.HTTPBadRequest(reason="'messages' field is required")
# Get PydanticAI agent with or without tools
# Get the agent instance
agent = get_pydantic_agent(discover_tools=enable_tools, user_id=user_id)
# Non-streaming response
# For non-streaming requests, collect the full response
if not stream:
response_content = await agent.chat_completion(
messages=messages,
conversation_id=conversation_id
)
return web.json_response({
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion",
"created": int(time.time()),
"model": "pydantic",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": response_content},
"message": {
"role": "assistant",
"content": response_content
},
"finish_reason": "stop"
}],
"model": model,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
@@ -252,182 +77,203 @@ async def chat_pydantic(request):
"tools_count": len(agent.tools_dict) if enable_tools else 0
})
else:
# Streaming response
# Handle streaming response
response = web.StreamResponse(
status=200,
headers={'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive'}
reason='OK',
headers={
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
}
)
await response.prepare(request)
async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True):
chunk_type = chunk.get("type", "content")
try:
async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True):
if chunk["type"] == "content":
chunk_data = {
"choices": [{
"index": 0,
"delta": {"content": chunk["content"]},
"finish_reason": chunk.get("finish_reason")
}],
"model": model
}
await response.write(f"data: {json.dumps(chunk_data)}\n\n".encode('utf-8'))
if chunk_type == "content":
json_chunk = {
"id": f"chatcmpl-{os.urandom(12).hex()}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": "pydantic",
"choices": [{
"index": 0,
"delta": {"content": chunk.get("content", "")},
"finish_reason": chunk.get("finish_reason")
}]
}
await response.write(f"data: {json.dumps(json_chunk)}\n\n".encode())
await response.write(b"data: [DONE]\n\n")
finally:
await response.write_eof()
if chunk.get("finish_reason") == "stop":
break
elif chunk_type == "error":
error_chunk = {
"error": {"message": chunk.get("content", "Unknown error")}
}
await response.write(f"data: {json.dumps(error_chunk)}\n\n".encode())
break
await response.write(b"data: [DONE]\n\n")
await response.write_eof()
return response
except web.HTTPBadRequest as e:
logger.warning(f"Bad request: {e.reason}")
return web.json_response({"error": {"message": e.reason}}, status=400)
except web.HTTPBadRequest:
raise
except Exception as e:
logger.exception("[PYDANTIC_AI] Error during chat completion:")
return web.json_response({"error": {"message": str(e)}}, status=500)
logger.exception(f"Error in chat_completions: {e}")
return web.json_response({
"error": {"message": f"Internal server error: {str(e)}"}
}, status=500)
async def list_models(request):
async def chat_simple(request):
"""
Lists available models (OpenAI-compatible endpoint).
Endpoint: GET /v1/models
"""
models = [
{
"id": "Tatlock",
"object": "model",
"created": int(time.time()),
"owned_by": "core-ai",
"permission": [],
"root": "tatlock",
"parent": None,
},
{
"id": "simple",
"object": "model",
"created": int(time.time()),
"owned_by": "core-ai",
"permission": [],
"root": "simple",
"parent": None,
}
]
return web.json_response({
"object": "list",
"data": models
})
async def list_tools(request):
"""
Lists all available tools.
Endpoint: GET /v1/tools
Handles chat requests using SimpleLiteLLMAgent (fallback, no tools).
Endpoint: /v1/chat/simple
"""
try:
data = await request.json()
logger.info(f"[SIMPLE/LITELLM] Received chat request")
# Extract relevant fields from the request
messages = data.get("messages")
model = data.get("model", "simple")
stream = data.get("stream", False)
conversation_id = data.get("conversation_id")
if not messages:
raise web.HTTPBadRequest(reason="'messages' field is required")
# Get simple agent instance (no tools)
agent = get_simple_litellm_agent()
# For non-streaming requests
if not stream:
response_content = await agent.chat_completion(
messages=messages,
conversation_id=conversation_id
)
return web.json_response({
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": response_content
},
"finish_reason": "stop"
}],
"model": model,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0
}
})
else:
# Streaming response
response = web.StreamResponse(
status=200,
reason='OK',
headers={
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
}
)
await response.prepare(request)
try:
async for chunk in agent.chat(messages=messages, conversation_id=conversation_id, stream=True):
if chunk["type"] == "content":
chunk_data = {
"choices": [{
"index": 0,
"delta": {"content": chunk["content"]},
"finish_reason": chunk.get("finish_reason")
}],
"model": model
}
await response.write(f"data: {json.dumps(chunk_data)}\n\n".encode('utf-8'))
await response.write(b"data: [DONE]\n\n")
finally:
await response.write_eof()
return response
except web.HTTPBadRequest:
raise
except Exception as e:
logger.exception(f"Error in chat_simple: {e}")
return web.json_response({
"error": {"message": f"Internal server error: {str(e)}"}
}, status=500)
async def list_models(request):
"""List available models"""
return web.json_response({
"object": "list",
"data": [
{
"id": "pydantic",
"object": "model",
"created": int(time.time()),
"owned_by": "core-ai"
},
{
"id": "simple",
"object": "model",
"created": int(time.time()),
"owned_by": "core-ai"
}
]
})
async def list_tools(request):
"""List all available tools"""
try:
tools = get_all_tools()
tool_list = []
tools_info = []
for name, func in tools.items():
tools_info.append({
import inspect
doc = inspect.getdoc(func) or "No description"
tool_list.append({
"name": name,
"description": func.__doc__.strip() if func.__doc__ else "No description available",
"type": "local"
"description": doc.split('\n')[0],
"type": "local" if not name.startswith("core-api__") else "openapi"
})
return web.json_response({
"tools": tools_info,
"count": len(tools_info),
"pydantic_ai_available": PYDANTIC_AI_AVAILABLE
"tools": tool_list,
"tools_count": len(tool_list)
})
except Exception as e:
logger.exception("Error listing tools:")
return web.json_response({"error": {"message": str(e)}}, status=500)
logger.exception(f"Error listing tools: {e}")
return web.json_response({
"error": {"message": str(e)}
}, status=500)
async def health_check(request):
"""Simple health check endpoint."""
"""Health check endpoint"""
return web.json_response({
"status": "ok",
"service": "core-ai",
"agents": {
"simple": True,
"ollama-native": OLLAMA_NATIVE_AVAILABLE,
"pydantic": PYDANTIC_AI_AVAILABLE
},
"default_agent": "ollama-native" if OLLAMA_NATIVE_AVAILABLE else "simple",
"default_agent": "pydantic" if PYDANTIC_AI_AVAILABLE else "simple",
"tools_count": len(get_all_tools())
})
async def test_ollama_tools(request):
"""Test Ollama tool calling directly"""
import httpx
try:
tool_def = {
"type": "function",
"function": {
"name": "web_search",
"description": "Search the web using SearXNG",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
}
}
payload = {
"model": "mistral-nemo:latest",
"messages": [{"role": "user", "content": "Search for Python 3.13 features"}],
"tools": [tool_def],
"stream": False
}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post('http://ollama:11434/api/chat', json=payload)
result = response.json()
return web.json_response({
"status_code": response.status_code,
"has_tool_calls": 'tool_calls' in result.get('message', {}),
"response": result
})
except Exception as e:
logger.exception("Test error:")
return web.json_response({"error": str(e)}, status=500)
async def setup_routes(app):
# Chat endpoints
app.router.add_post("/chat/completions", chat_completions) # Alias without /v1 for compatibility
app.router.add_post("/v1/chat/completions", chat_completions) # Default (PydanticAI)
app.router.add_post("/v1/chat/simple", chat_simple) # Simple agent (no tools)
app.router.add_post("/v1/chat/pydantic", chat_pydantic) # Alias for default
app.router.add_post("/chat/completions", chat_completions) # Alias without /v1
app.router.add_post("/v1/chat/completions", chat_completions) # Default: PydanticAI
app.router.add_post("/v1/chat/simple", chat_simple) # Fallback: Simple agent
# OpenAI-compatible endpoints
app.router.add_get("/v1/models", list_models) # List available models
app.router.add_get("/models", list_models) # Alias without /v1 prefix
# Models endpoint
app.router.add_get("/models", list_models)
app.router.add_get("/v1/models", list_models)
# Tool management
app.router.add_get("/v1/tools", list_tools) # List available tools
# Tools endpoint
app.router.add_get("/tools", list_tools)
app.router.add_get("/v1/tools", list_tools)
# Health check
app.router.add_get("/health", health_check)
app.router.add_get("/test/ollama-tools", test_ollama_tools)
# Setup CORS
cors = cors_setup(app, defaults={
@@ -439,20 +285,21 @@ async def setup_routes(app):
)
})
# Configure CORS on all routes
# Apply CORS to all routes
for route in list(app.router.routes()):
cors.add(route)
def main():
app = web.Application()
app.on_startup.append(setup_routes) # Register routes on startup
# Configuration
host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", 8086)) # Use 8086 to avoid conflict with core-ai
logger.info(f"Starting core-ai service on http://{host}:{port}")
web.run_app(app, host=host, port=port)
if __name__ == "__main__":
main()
# Set up routes
import asyncio
loop = asyncio.get_event_loop()
loop.run_until_complete(setup_routes(app))
# Run the application
logger.info("Starting core-ai service on http://0.0.0.0:8086")
web.run_app(app, host='0.0.0.0', port=8086)
if __name__ == '__main__':
main()
services/core-ai/src/agents/__init__.py
-6
View File
@@ -1,9 +1,6 @@
"""Agent implementations for core-ai service"""
from .simple import SimpleLiteLLMAgent, get_simple_litellm_agent
from .ollama_native_agent import OllamaNativeAgent, get_ollama_native_agent
OLLAMA_NATIVE_AVAILABLE = True
try:
from .pydantic_agent import PydanticAgent, get_pydantic_agent
@@ -16,9 +13,6 @@ except ImportError:
__all__ = [
'SimpleLiteLLMAgent',
'get_simple_litellm_agent',
'OllamaNativeAgent',
'get_ollama_native_agent',
'OLLAMA_NATIVE_AVAILABLE',
'PydanticAgent',
'get_pydantic_agent',
'PYDANTIC_AI_AVAILABLE',
services/core-ai/src/agents/ollama_native_agent.py
-297
View File
@@ -1,297 +0,0 @@
"""
Native Ollama Agent - Uses Ollama's native API with tool calling support.
This agent bypasses PydanticAI's OpenAI-compatible approach and uses
Ollama's native /api/chat endpoint which has better tool calling support.
"""
import logging
import httpx
import json
from typing import List, Dict, Any, AsyncIterator
from functools import lru_cache
from src.config import get_settings
from src.prompts import get_prompt
from src.tools.registry import get_all_tools
logger = logging.getLogger(__name__)
class OllamaNativeAgent:
"""
Agent using Ollama's native API with tool calling support.
Unlike PydanticAI which uses Ollama's OpenAI-compatible API,
this uses the native /api/chat endpoint which has proper tool support.
"""
def __init__(self, tools: List = None, discover_tools: bool = False, include_openapi: bool = True):
logger.info("OllamaNativeAgent: Initializing...")
self.settings = get_settings()
self.model = self.settings.agent_model
self.include_openapi = include_openapi
self._tools_loaded = False
# Load system prompt
from datetime import datetime
base_prompt = get_prompt("pydantic_agent")
current_date = datetime.now().strftime("%A, %B %d, %Y")
self.system_prompt = f"Today is {current_date}.\n\n{base_prompt}"
# Get tools (sync part only)
if tools is not None:
self.tools_dict = {func.__name__: func for func in tools}
self._tools_loaded = True
elif discover_tools:
# Get core tools (local) - sync
self.tools_dict = get_all_tools()
# OpenAPI tools will be loaded async on first use
else:
self.tools_dict = {}
self._tools_loaded = True
logger.info(f"OllamaNativeAgent: {len(self.tools_dict)} core tools loaded")
logger.info(f"OllamaNativeAgent: Model: {self.model}")
logger.info("✓ OllamaNativeAgent: Initialization complete")
async def _ensure_tools_loaded(self):
"""Load OpenAPI tools asynchronously (called on first use)"""
if self._tools_loaded:
return
if self.include_openapi and self.settings.openapi_enabled:
try:
from src.tools.openapi_discovery import get_openapi_tools
# Parse OpenAPI endpoints from config
endpoints = [e.strip() for e in self.settings.openapi_endpoints.split(",")]
# Fetch OpenAPI tools (async)
openapi_tools = await get_openapi_tools(endpoints=endpoints)
self.tools_dict.update(openapi_tools)
logger.info(f"OllamaNativeAgent: Added {len(openapi_tools)} OpenAPI tools")
except Exception as e:
logger.warning(f"OllamaNativeAgent: Failed to load OpenAPI tools: {e}")
self._tools_loaded = True
logger.info(f"OllamaNativeAgent: Total tools available: {len(self.tools_dict)}")
def _format_tools_for_ollama(self) -> List[Dict[str, Any]]:
"""
Convert Python functions to Ollama tool format.
Ollama expects:
{
"type": "function",
"function": {
"name": "function_name",
"description": "...",
"parameters": {...JSON Schema...}
}
}
"""
tools = []
for name, func in self.tools_dict.items():
# Extract function signature and docstring
import inspect
sig = inspect.signature(func)
doc = inspect.getdoc(func) or "No description"
# Build parameters schema
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name in ['self', 'cls']:
continue
# Determine type
param_type = "string" # default
if param.annotation != inspect.Parameter.empty:
if param.annotation == int:
param_type = "integer"
elif param.annotation == float:
param_type = "number"
elif param.annotation == bool:
param_type = "boolean"
properties[param_name] = {
"type": param_type,
"description": f"Parameter {param_name}"
}
# Required if no default value
if param.default == inspect.Parameter.empty:
required.append(param_name)
tool_def = {
"type": "function",
"function": {
"name": name,
"description": doc.split('\n')[0], # First line of docstring
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
tools.append(tool_def)
return tools
async def chat(
self,
messages: List[Dict[str, str]],
conversation_id: str = None,
stream: bool = True
) -> AsyncIterator[Dict[str, Any]]:
"""
Process chat messages with tool calling support.
Args:
messages: List of message dicts with 'role' and 'content'
conversation_id: Optional conversation ID
stream: Whether to stream responses
Yields:
Dict with 'type' and content
"""
# Ensure OpenAPI tools are loaded (async, called once)
await self._ensure_tools_loaded()
logger.info(f"OllamaNativeAgent: Processing message: {messages[-1]['content'][:50]}...")
try:
# Extract user message
user_messages = [m for m in messages if m["role"] != "system"]
if not user_messages:
raise ValueError("No user messages provided")
# Build Ollama messages format
ollama_messages = [
{"role": "system", "content": self.system_prompt}
]
ollama_messages.extend(user_messages)
# Format tools
tools = self._format_tools_for_ollama() if self.tools_dict else None
# Make request to Ollama
payload = {
"model": self.model,
"messages": ollama_messages,
"stream": False # Handle streaming separately if needed
}
if tools:
payload["tools"] = tools
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.settings.ollama_base_url}/api/chat",
json=payload
)
response.raise_for_status()
result = response.json()
message = result.get("message", {})
# Check if model wants to call tools
if "tool_calls" in message and message["tool_calls"]:
logger.info(f"Tool calls requested: {len(message['tool_calls'])}")
# Execute tools
tool_results = []
for tool_call in message["tool_calls"]:
func_name = tool_call["function"]["name"]
func_args = tool_call["function"]["arguments"]
logger.info(f"Executing tool: {func_name}({func_args})")
if func_name in self.tools_dict:
try:
tool_func = self.tools_dict[func_name]
# Call tool (handle both sync and async)
import asyncio
if asyncio.iscoroutinefunction(tool_func):
tool_result = await tool_func(**func_args)
else:
tool_result = tool_func(**func_args)
tool_results.append({
"role": "tool",
"content": str(tool_result)
})
logger.info(f"Tool result: {str(tool_result)[:100]}...")
except Exception as e:
error_msg = f"Tool execution error: {str(e)}"
logger.error(error_msg)
tool_results.append({
"role": "tool",
"content": error_msg
})
else:
logger.warning(f"Tool {func_name} not found")
tool_results.append({
"role": "tool",
"content": f"Error: Tool {func_name} not available"
})
# Send tool results back to model
ollama_messages.append(message)
ollama_messages.extend(tool_results)
payload["messages"] = ollama_messages
payload.pop("tools", None) # Don't send tools again
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.settings.ollama_base_url}/api/chat",
json=payload
)
response.raise_for_status()
final_result = response.json()
final_content = final_result.get("message", {}).get("content", "")
logger.info(f"Final response: {final_content[:100]}...")
yield {"type": "content", "content": final_content, "finish_reason": "stop"}
else:
# No tool calls, return response directly
content = message.get("content", "")
logger.info(f"Direct response: {content[:100]}...")
yield {"type": "content", "content": content, "finish_reason": "stop"}
except Exception as e:
logger.error(f"OllamaNativeAgent error: {e}", exc_info=True)
yield {
"type": "error",
"content": f"Error: {str(e)}",
"finish_reason": "error"
}
async def chat_completion(
self,
messages: List[Dict[str, str]],
conversation_id: str = None
) -> str:
"""Non-streaming chat completion."""
final_content = ""
async for chunk in self.chat(messages=messages, conversation_id=conversation_id, stream=False):
if chunk["type"] == "content":
final_content += chunk["content"]
return final_content if final_content else "I couldn't generate a response."
@lru_cache()
def get_ollama_native_agent(discover_tools: bool = True) -> OllamaNativeAgent:
"""Get cached Ollama native agent instance."""
return OllamaNativeAgent(discover_tools=discover_tools)
services/core-ai/src/prompts.py
+26 -9
View File
@@ -13,18 +13,35 @@ Your core responsibility: Verify facts before presenting them as truth.
You have access to two categories of tools:
**Core Tools** (essential utilities):
- web_search: Latest/current/recent information (always verify facts, sir)
**Core Tools** (always available):
- web_search: ONLY for current events, news, research, or information not available through other tools
- calculate: Mathematical operations (precision is paramount)
- get_current_time/get_current_date: Time/date queries
- add_days_to_date/calculate_date_difference: Date calculations
**Infrastructure Tools** (prefixed with "core_api__"):
When managing sir's home infrastructure, use these tools:
- Services: List, start, stop Docker services
- Domains: List configured domains
- Proxy: Manage reverse proxy configurations
- Monitors: Health monitoring (Uptime Kuma integration)
- Ports: Check allocated ports
**Infrastructure Tools** (discovered from core-api, prefixed with "core_api__"):
These tools provide DIRECT system access. PREFER these over web searches when available:
- DNS queries: core_api__dns_lookup_tools_dns_lookup_post
* Use for: A, AAAA, MX, TXT, CNAME, NS, SOA, PTR records
* Example: "lookup A records for github.com" → use DNS tool, NOT web_search
- Web scraping: core_api__scrape_website_tools_scrape_post
* Use for: Extract content from specific URLs
- Docker services: core_api__list_services*, core_api__get_service*, core_api__start_service*, core_api__stop_service*
* Use for: Manage sir's containerized services
- Network management: core_api__list_domains*, core_api__list_ports*, core_api__get_proxy_host*
* Use for: Infrastructure configuration
- Monitoring: core_api__list_monitors*, core_api__get_monitor*, core_api__create_monitor*
* Use for: System health checks
**Tool Selection Priority:**
1. If an infrastructure tool exists for the task → use it (more reliable than web search)
2. If no infrastructure tool exists → use web_search
3. For general knowledge → answer directly (no tool needed)
Be concise unless details are specifically requested. When using tools, acknowledge them naturally in your dignified manner."""
}
services/core-ai/src/tools.py
-101
View File
@@ -1,101 +0,0 @@
"""
Agent Tools - Tools for the Core AI agent
These tools make REST API calls to the Core API service.
"""
from typing import List, Dict, Any
import logging
import functools
import inspect
import httpx # For making asynchronous HTTP requests
# Adjusted import path for the new core-ai service structure
from src.config import get_settings
logger = logging.getLogger(__name__)
# Initialize settings once
settings = get_settings()
CORE_API_BASE_URL = settings.core_api_base_url
# ============================================================================
# Decorator for logging tool calls
# ============================================================================
def log_tool_call(func):
"""Decorator to log tool calls with their parameters"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
params_str = ", ".join(
[f"{arg}" for arg in args] +
[f"{k}={repr(v)}" for k, v in kwargs.items()]
)
logger.info(f"🔧 TOOL CALL: {func.__name__}({params_str})")
try:
sig = inspect.signature(func)
valid_kwargs = {
key: value for key, value in kwargs.items()
if key in sig.parameters
}
result = await func(*args, **valid_kwargs)
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}", exc_info=True)
raise
return wrapper
# ============================================================================
# HTTP Client
# ============================================================================
# Use a single httpx client for performance
# It's important to close the client when the application shuts down
http_client = httpx.AsyncClient()
# ============================================================================
# Knowledge & Search Tools
# ============================================================================
@log_tool_call
async def web_search(query: str, num_results: int) -> str:
"""
Search the web. (Neutered for testing purposes).
"""
logger.info(f"--- NEUTERED WEB SEARCH CALLED FOR: {query} ---")
if "capital of france" in query.lower():
return "Search results for 'Capital of France':\n\n1. **Paris - Wikipedia**\n URL: https://en.wikipedia.org/wiki/Paris\n Paris is the capital and most populous city of France."
else:
return f"Search results for '{query}':\n\n1. No specific results for this neutered test. Try 'capital of France'."
# ============================================================================
# Special Tools (Response tool is here for consistency, but will be removed for initial test)
# ============================================================================
@log_tool_call
async def response(answer: str) -> None:
"""
Deliver your final response to the user.
"""
logger.info("`response` tool called. Returning None to terminate agent loop.")
return None
# ============================================================================
# Tool Registry - Legacy (deprecated, use src/tools/registry.py instead)
# ============================================================================
try:
from google.adk.tools import FunctionTool
ADK_AVAILABLE = True
except ImportError:
ADK_AVAILABLE = False
FunctionTool = None
def get_agent_tools() -> List:
"""DEPRECATED: Get all tools available to the agent. Use src/tools/registry.py instead."""
logger.info("--- DIAGNOSTIC MODE (Phase 1): Agent has NO tools. ---")
return []
services/core-ai/tests/QUALITY_TESTS.md
-370
View File
@@ -1,370 +0,0 @@
# Core-AI Quality Test Suite
Comprehensive test suite for benchmarking AI agent performance and detecting regressions across code changes.
## Purpose
This test suite validates:
- **Tool calling decision-making** - Does the agent choose the right tools?
- **Response quality** - Are responses accurate and complete?
- **Performance** - Are responses delivered within acceptable timeframes?
- **Regression detection** - Has quality degraded since the last version?
## Current Implementation
- **Agent**: OllamaNativeAgent (Ollama native API with tool calling)
- **Model**: mistral-nemo:latest
- **Framework**: PydanticAI
- **Tools**: web_search, calculate, date/time operations
## Quick Start
### Run All Tests
```bash
# From services/core-ai directory
python tests/test_ai_flow_quality.py
```
This will:
1. Run all test scenarios
2. Generate a comprehensive report
3. Save reports to `tests/reports/` with timestamp and git tag
4. Output results to console
### Run with pytest
```bash
# Run all tests
pytest tests/test_ai_flow_quality.py -v
# Run specific scenario
pytest tests/test_ai_flow_quality.py::test_scenario2_web_search -v
# Run regression tests only
pytest tests/test_ai_flow_quality.py -k regression -v
```
## Test Scenarios
### Scenario 1: Simple Knowledge Query
- **Query**: "What is Docker?"
- **Expected**: Direct answer without tools
- **Performance Target**: < 10s
### Scenario 2: Web Search
- **Query**: "What are the latest Kubernetes security best practices?"
- **Expected**: Uses web_search tool, synthesizes results
- **Performance Target**: < 30s
### Scenario 3: Mathematical Calculation
- **Query**: "Calculate 2847 * 1923 + 5612 - 999"
- **Expected**: Uses calculate tool for precision
- **Performance Target**: < 15s
- **Expected Result**: 5,479,394
### Scenario 4: Date/Time Operations
- **Query**: "What is the current date and what will it be in 30 days?"
- **Expected**: Uses get_current_date and add_days_to_date tools
- **Performance Target**: < 15s
### Scenario 5: Multi-Tool Complex Query
- **Query**: "Get current time in NYC and Tokyo, calculate difference"
- **Expected**: Multiple tool calls (get_current_time × 2, synthesis)
- **Performance Target**: < 30s
### Scenario 6: DNS Lookup (OpenAPI Discovery)
- **Query**: "What are the A records for github.com? Use the core-api DNS lookup tool."
- **Expected**: Uses DNS lookup tool discovered via OpenAPI from core-api
- **Performance Target**: < 15s
- **Purpose**: Tests OpenAPI tool discovery and infrastructure integration
## Report Format
Reports are saved in two formats:
### 1. Text Report (`quality-report-YYYYMMDD-HHMMSS.txt`)
```
================================================================================
CORE-AI QUALITY REPORT
Generated: 2025-12-01T10:30:45
Git Tag: v2.1.0
Git Commit: a3b2c1d
Git Branch: main
Implementation:
Agent: OllamaNativeAgent
Model: mistral-nemo:latest
Framework: PydanticAI
API: Ollama native (/api/chat)
================================================================================
Total Tests: 5
Passed: 5 (100.0%)
Failed: 0
Performance:
Average response time: 8.45s
Fastest response: 3.21s
Slowest response: 15.67s
Test Details:
--------------------------------------------------------------------------------
1. Scenario 1: Simple Knowledge: ✓ PASS
Query: What is Docker?...
Response time: 3.21s
Tools available: 6
Response length: 245 chars
...
================================================================================
REVERT INSTRUCTIONS:
If quality has degraded, revert to: v2.1.0
git checkout v2.1.0
================================================================================
```
### 2. JSON Report (`quality-report-YYYYMMDD-HHMMSS.json`)
Machine-readable format for programmatic analysis and trend tracking:
```json
{
"timestamp": "2025-12-01T10:30:45",
"git_info": {
"tag": "v2.1.0",
"commit": "a3b2c1d",
"branch": "main"
},
"summary": {
"total": 5,
"passed": 5,
"failed": 0
},
"results": [...]
}
```
## Workflow: Before Making Changes
### 1. Establish Baseline
Before making any code changes, run the test suite to establish a quality baseline:
```bash
cd /home/jpmschweitzer/Projects/portainer-core/services/core-ai
python tests/test_ai_flow_quality.py
```
**Save the report location** - you'll compare against this later.
### 2. Make Your Changes
Edit agent code, tools, prompts, etc.
### 3. Run Tests Again
```bash
python tests/test_ai_flow_quality.py
```
### 4. Compare Reports
Compare the new report against the baseline:
```bash
# List recent reports
ls -lh tests/reports/
# Compare two reports
diff tests/reports/quality-report-20251201-103045.txt \
tests/reports/quality-report-20251201-115522.txt
```
**Key metrics to watch**:
- Pass rate (should stay 100%)
- Average response time (should not significantly increase)
- Individual test failures (investigate immediately)
### 5. Revert if Quality Degrades
If tests fail or performance degrades significantly:
```bash
# Check the git tag from the failing report
cat tests/reports/quality-report-20251201-115522.txt | grep "Git Tag"
# Revert to that tag
git checkout v2.1.0
```
## Benchmarking Models
To compare different models:
### 1. Run baseline with current model
```bash
python tests/test_ai_flow_quality.py
# Save this as baseline
```
### 2. Change model in config
Edit `services/core-ai/src/config.py` or environment variable:
```python
# Change from mistral-nemo:latest to gemma2:9b
AGENT_MODEL = "gemma2:9b"
```
Restart core-ai:
```bash
cd /home/jpmschweitzer/Projects/portainer-core/stacks
docker restart core-ai
```
### 3. Run tests with new model
```bash
python tests/test_ai_flow_quality.py
```
### 4. Compare results
```bash
# Check both JSON reports for performance comparison
cat tests/reports/quality-report-BASELINE.json | jq '.summary'
cat tests/reports/quality-report-NEW_MODEL.json | jq '.summary'
```
Look for:
- **Pass rate changes** - Did the new model fail any tests?
- **Response time changes** - Is it faster or slower?
- **Response quality** - Are answers as good?
## Troubleshooting
### Tests Fail: "Connection refused"
**Problem**: core-ai service not running
**Solution**:
```bash
cd /home/jpmschweitzer/Projects/portainer-core/stacks
docker restart core-ai
docker logs core-ai # Check for startup errors
```
### Tests Timeout
**Problem**: Model too slow or stuck
**Solution**:
1. Check Ollama GPU usage: `nvidia-smi`
2. Check model is loaded: `docker exec ollama ollama list`
3. Increase timeout in test file if needed
### Web Search Tests Fail
**Problem**: SearXNG not available
**Solution**:
```bash
docker restart searxng
curl "http://localhost:8087/search?q=test&format=json"
```
### Calculation Tests Fail
**Problem**: Agent not using calculate tool
**Solution**: Check tool registration:
```bash
curl http://localhost:8086/v1/tools | jq '.tools[] | .name'
```
## Adding New Test Scenarios
### 1. Add test function
```python
@pytest.mark.asyncio
async def test_my_new_scenario():
"""
Scenario: My New Feature
Expected: Describe expected behavior
Performance target: < Xs
"""
async with AIFlowTester() as tester:
result = await tester.chat("My test query")
tester.assert_response_quality(
result,
expected_keywords=["keyword1", "keyword2"],
min_length=50,
max_time=15.0
)
# Custom assertions
assert "expected result" in result["response"]
print(f"✓ My scenario: {result['total_time']:.2f}s")
```
### 2. Add to scenario list
In `run_full_quality_check()`:
```python
test_scenarios = [
# ... existing scenarios ...
{
"name": "Scenario X: My New Feature",
"query": "My test query",
"test": test_my_new_scenario
},
]
```
### 3. Run to verify
```bash
pytest tests/test_ai_flow_quality.py::test_my_new_scenario -v
```
## Best Practices
### ✅ DO:
- Run tests before committing major changes
- Compare reports to detect regressions
- Save baseline reports for each release
- Document expected behavior in test docstrings
- Use meaningful git tags for easy reversion
### ❌ DON'T:
- Skip tests when making agent changes
- Ignore performance degradation warnings
- Delete old reports (keep for trend analysis)
- Change test expectations to make tests pass
- Commit without running tests first
## Report Retention
Keep reports organized:
```bash
# Keep last 30 days of reports
find tests/reports/ -name "*.txt" -mtime +30 -delete
find tests/reports/ -name "*.json" -mtime +30 -delete
# Archive reports by month
mkdir -p tests/reports/archive/2025-12/
mv tests/reports/quality-report-202512*.* tests/reports/archive/2025-12/
```
---
**Remember**: These tests protect quality. If they fail, investigate before proceeding!
services/core-ai/tests/test_02_litellm_raw.py
-137
View File
@@ -1,137 +0,0 @@
#!/usr/bin/env python3
"""
Layer 2: Raw LiteLLM Connection Tests
Tests direct LiteLLM → Ollama communication without any wrappers.
"""
import pytest
import sys
from pathlib import Path
import time
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.config import get_settings
try:
import litellm
except ImportError:
pytest.skip("LiteLLM not installed", allow_module_level=True)
@pytest.mark.asyncio
async def test_litellm_simple_completion():
"""Test a simple LiteLLM completion"""
settings = get_settings()
model = f"ollama/{settings.agent_model}"
messages = [{"role": "user", "content": "Say 'test' and nothing else."}]
print(f"\n→ Testing LiteLLM with model: {model}")
print(f"→ API base: {settings.ollama_base_url}")
start_time = time.time()
response = await litellm.acompletion(
model=model,
messages=messages,
api_base=settings.ollama_base_url,
temperature=0.1,
stream=False
)
elapsed = time.time() - start_time
content = response.choices[0].message.content
assert content is not None, "No content in response"
assert len(content) > 0, "Empty content"
print(f"✓ Response received in {elapsed:.2f}s: {content}")
@pytest.mark.asyncio
async def test_litellm_with_system_prompt():
"""Test LiteLLM completion with system prompt"""
settings = get_settings()
model = f"ollama/{settings.agent_model}"
messages = [
{"role": "system", "content": "You are a helpful assistant. Be concise."},
{"role": "user", "content": "What is 2+2? Answer with just the number."}
]
response = await litellm.acompletion(
model=model,
messages=messages,
api_base=settings.ollama_base_url,
temperature=0.1,
stream=False
)
content = response.choices[0].message.content
assert content is not None, "No content in response"
print(f"✓ Response with system prompt: {content}")
@pytest.mark.asyncio
async def test_litellm_streaming():
"""Test LiteLLM streaming mode"""
settings = get_settings()
model = f"ollama/{settings.agent_model}"
messages = [{"role": "user", "content": "Count from 1 to 3. Just numbers."}]
response = await litellm.acompletion(
model=model,
messages=messages,
api_base=settings.ollama_base_url,
temperature=0.1,
stream=True
)
chunks = []
chunk_count = 0
async for chunk in response:
chunk_count += 1
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
full_content = "".join(chunks)
assert chunk_count > 0, "No chunks received"
assert len(full_content) > 0, "No content in chunks"
print(f"✓ Received {chunk_count} chunks: {full_content}")
@pytest.mark.asyncio
async def test_litellm_capital_of_france():
"""Test the actual failing case: 'What is the capital of France?'"""
settings = get_settings()
model = f"ollama/{settings.agent_model}"
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is the capital of France?"}
]
print(f"\n→ Testing the actual failing query...")
response = await litellm.acompletion(
model=model,
messages=messages,
api_base=settings.ollama_base_url,
temperature=0.1,
stream=False
)
content = response.choices[0].message.content
assert content is not None, "No content in response"
assert len(content) > 0, "Empty response"
# Check if the answer is reasonable
content_lower = content.lower()
assert "paris" in content_lower, f"Expected 'Paris' in answer, got: {content}"
print(f"✓ Correct answer received: {content}")
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])
services/core-ai/tests/test_03_message_format.py
-113
View File
@@ -1,113 +0,0 @@
#!/usr/bin/env python3
"""
Layer 3: Message Formatting & Prompts Tests
Tests that system prompts are correctly injected and messages are formatted properly.
"""
import pytest
import sys
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.config import get_settings
from src.prompts import get_prompt, PROMPTS
def test_prompts_defined():
"""Test that prompts are defined"""
assert len(PROMPTS) > 0, "No prompts defined"
print(f"✓ {len(PROMPTS)} prompt variant(s) defined")
def test_get_prompt_default():
"""Test getting default prompt"""
prompt = get_prompt()
assert prompt is not None, "Default prompt is None"
assert len(prompt) > 0, "Default prompt is empty"
print(f"✓ Default prompt: {prompt[:80]}...")
def test_get_prompt_specific():
"""Test getting specific prompt variant"""
settings = get_settings()
prompt = get_prompt(settings.system_prompt_variant)
assert prompt is not None, f"Prompt '{settings.system_prompt_variant}' is None"
assert len(prompt) > 0, f"Prompt '{settings.system_prompt_variant}' is empty"
print(f"✓ Prompt '{settings.system_prompt_variant}': {prompt[:80]}...")
def test_message_structure():
"""Test that message structure is valid"""
test_messages = [
{"role": "user", "content": "Hello"}
]
# Simulate system prompt injection
from src.prompts import get_prompt
system_prompt = get_prompt()
formatted_messages = [
{"role": "system", "content": system_prompt},
*test_messages
]
# Validate structure
assert len(formatted_messages) == 2, "Expected 2 messages after injection"
assert formatted_messages[0]["role"] == "system", "First message should be system"
assert formatted_messages[1]["role"] == "user", "Second message should be user"
print(f"✓ Message structure correct")
for i, msg in enumerate(formatted_messages):
content_preview = msg['content'][:50] + "..." if len(msg['content']) > 50 else msg['content']
print(f" [{i}] {msg['role']}: {content_preview}")
def test_system_prompt_not_duplicated():
"""Test that system prompt is not duplicated if already present"""
from src.prompts import get_prompt
system_prompt = get_prompt()
# Messages already have system prompt
messages = [
{"role": "system", "content": "Custom system prompt"},
{"role": "user", "content": "Hello"}
]
# Simulate the check in agent
if messages and messages[0]["role"] == "system":
# Should not inject
formatted_messages = messages
else:
# Would inject
formatted_messages = [{"role": "system", "content": system_prompt}, *messages]
# Should still have only 2 messages
assert len(formatted_messages) == 2, "System prompt was duplicated"
assert formatted_messages[0]["role"] == "system"
assert formatted_messages[0]["content"] == "Custom system prompt"
print(f"✓ System prompt not duplicated when already present")
def test_empty_messages_handling():
"""Test handling of empty messages list"""
from src.prompts import get_prompt
system_prompt = get_prompt()
messages = []
# Simulate injection
if not messages or messages[0]["role"] != "system":
formatted_messages = [{"role": "system", "content": system_prompt}, *messages]
else:
formatted_messages = messages
assert len(formatted_messages) == 1, "Should have system prompt only"
assert formatted_messages[0]["role"] == "system"
print(f"✓ Empty messages handled correctly")
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])
services/core-ai/tests/test_ai_flow_quality.py
-634
View File
@@ -1,634 +0,0 @@
"""
AI Flow Quality Test Suite
Purpose: Benchmark core-ai agent behavior for regression detection
and performance tracking over time.
This test suite validates:
- Tool calling decision-making (OllamaNativeAgent)
- Multi-step reasoning capability
- Response quality and accuracy
- Performance characteristics
- Regression detection across code changes
Current Implementation:
- Agent: OllamaNativeAgent (Ollama native API with tool calling)
- Model: mistral-nemo:latest (primary reasoning model)
- Tools: Local tools (web_search, calculate, date/time operations)
- Framework: PydanticAI
Usage:
pytest tests/test_ai_flow_quality.py -v
pytest tests/test_ai_flow_quality.py::test_simple_knowledge -v
python tests/test_ai_flow_quality.py # Run and generate report
Reports saved to: tests/reports/
"""
import asyncio
import time
import json
import httpx
import subprocess
from pathlib import Path
from typing import Dict, List, Any, Optional
from datetime import datetime
import pytest
# Test Configuration
BASE_URL = "http://localhost:8086"
TIMEOUT = 60.0
REPORTS_DIR = Path(__file__).parent / "reports"
def get_git_info() -> Dict[str, str]:
"""Get current git tag and commit hash"""
try:
# Get current tag
tag = subprocess.check_output(
["git", "describe", "--tags", "--exact-match"],
stderr=subprocess.DEVNULL,
text=True
).strip()
except subprocess.CalledProcessError:
# No exact tag, get latest tag + commit
try:
tag = subprocess.check_output(
["git", "describe", "--tags", "--always"],
text=True
).strip()
except subprocess.CalledProcessError:
tag = "unknown"
try:
commit = subprocess.check_output(
["git", "rev-parse", "--short", "HEAD"],
text=True
).strip()
except subprocess.CalledProcessError:
commit = "unknown"
try:
branch = subprocess.check_output(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
text=True
).strip()
except subprocess.CalledProcessError:
branch = "unknown"
return {"tag": tag, "commit": commit, "branch": branch}
class AIFlowTester:
"""Test harness for AI flow quality checks"""
def __init__(self):
self.results = []
self.client = None
self.git_info = get_git_info()
async def __aenter__(self):
self.client = httpx.AsyncClient(timeout=TIMEOUT)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.client:
await self.client.aclose()
async def chat(
self,
message: str,
enable_tools: bool = True,
stream: bool = False
) -> Dict[str, Any]:
"""
Send a chat request and measure performance
Returns:
{
"response": str,
"time_to_first_token": float,
"total_time": float,
"tools_count": int,
"success": bool,
"error": str | None
}
"""
start_time = time.time()
time_to_first_token = None
response_text = ""
try:
payload = {
"messages": [{"role": "user", "content": message}],
"stream": stream,
"enable_tools": enable_tools
}
response = await self.client.post(
f"{BASE_URL}/v1/chat/completions",
json=payload
)
response.raise_for_status()
if not stream:
time_to_first_token = time.time() - start_time
data = response.json()
response_text = data["choices"][0]["message"]["content"]
tools_count = data.get("tools_count", 0)
tools_enabled = data.get("tools_enabled", False)
else:
# Handle streaming response
# TODO: Implement streaming support
raise NotImplementedError("Streaming not yet implemented")
total_time = time.time() - start_time
return {
"response": response_text,
"time_to_first_token": time_to_first_token,
"total_time": total_time,
"tools_count": tools_count if tools_enabled else 0,
"success": True,
"error": None
}
except Exception as e:
total_time = time.time() - start_time
return {
"response": "",
"time_to_first_token": None,
"total_time": total_time,
"tools_count": 0,
"success": False,
"error": str(e)
}
def assert_response_quality(
self,
result: Dict[str, Any],
expected_keywords: List[str] = None,
min_length: int = 20,
max_time: float = 30.0
):
"""
Validate response quality
Args:
result: Test result from chat()
expected_keywords: Keywords that should appear in response
min_length: Minimum response length
max_time: Maximum acceptable response time
"""
assert result["success"], f"Request failed: {result['error']}"
assert len(result["response"]) >= min_length, \
f"Response too short: {len(result['response'])} < {min_length}"
assert result["total_time"] <= max_time, \
f"Response too slow: {result['total_time']:.2f}s > {max_time}s"
if expected_keywords:
response_lower = result["response"].lower()
for keyword in expected_keywords:
assert keyword.lower() in response_lower, \
f"Missing keyword '{keyword}' in response"
def generate_report(self) -> str:
"""Generate a quality report from test results"""
if not self.results:
return "No test results to report"
report = []
report.append("=" * 80)
report.append("CORE-AI QUALITY REPORT")
report.append(f"Generated: {datetime.now().isoformat()}")
report.append(f"Git Tag: {self.git_info['tag']}")
report.append(f"Git Commit: {self.git_info['commit']}")
report.append(f"Git Branch: {self.git_info['branch']}")
report.append("")
report.append("Implementation:")
report.append(" Agent: OllamaNativeAgent")
report.append(" Model: mistral-nemo:latest")
report.append(" Framework: PydanticAI")
report.append(" API: Ollama native (/api/chat)")
report.append("=" * 80)
report.append("")
# Summary statistics
total_tests = len(self.results)
passed_tests = sum(1 for r in self.results if r.get("passed", False))
failed_tests = total_tests - passed_tests
report.append(f"Total Tests: {total_tests}")
report.append(f"Passed: {passed_tests} ({100*passed_tests/total_tests:.1f}%)")
report.append(f"Failed: {failed_tests}")
report.append("")
# Performance metrics
response_times = [r["result"]["total_time"] for r in self.results
if r["result"]["success"]]
if response_times:
avg_time = sum(response_times) / len(response_times)
min_time = min(response_times)
max_time = max(response_times)
report.append("Performance:")
report.append(f" Average response time: {avg_time:.2f}s")
report.append(f" Fastest response: {min_time:.2f}s")
report.append(f" Slowest response: {max_time:.2f}s")
report.append("")
# Individual test results
report.append("Test Details:")
report.append("-" * 80)
for i, test in enumerate(self.results, 1):
status = "✓ PASS" if test.get("passed", False) else "✗ FAIL"
report.append(f"\n{i}. {test['name']}: {status}")
report.append(f" Query: {test['query'][:60]}...")
result = test["result"]
if result["success"]:
report.append(f" Response time: {result['total_time']:.2f}s")
report.append(f" Tools available: {result['tools_count']}")
report.append(f" Response length: {len(result['response'])} chars")
if test.get("error"):
report.append(f" ⚠ Assertion failed: {test['error']}")
else:
report.append(f" ✗ Error: {result['error']}")
report.append("")
report.append("=" * 80)
report.append("REVERT INSTRUCTIONS:")
report.append(f"If quality has degraded, revert to: {self.git_info['tag']}")
report.append(f" git checkout {self.git_info['tag']}")
report.append("=" * 80)
return "\n".join(report)
# ============================================================================
# TEST SCENARIOS
# ============================================================================
@pytest.mark.asyncio
async def test_scenario1_simple_knowledge():
"""
Scenario 1: Simple Knowledge Query (No Tools Needed)
Expected: Direct answer without requiring tools
Performance target: < 10s
"""
async with AIFlowTester() as tester:
result = await tester.chat("What is Docker?")
tester.assert_response_quality(
result,
expected_keywords=["container", "platform"],
min_length=50,
max_time=10.0
)
assert result["success"], "Request failed"
print(f"✓ Simple knowledge query: {result['total_time']:.2f}s")
@pytest.mark.asyncio
async def test_scenario2_web_search():
"""
Scenario 2: Web Search Required
Expected: Uses web_search tool via SearXNG, synthesizes results
Performance target: < 30s
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"What are the latest Kubernetes security best practices?"
)
tester.assert_response_quality(
result,
expected_keywords=["kubernetes", "security"],
min_length=100,
max_time=30.0
)
assert result["success"], "Request failed"
print(f"✓ Web search query: {result['total_time']:.2f}s")
print(f" Tools available: {result['tools_count']}")
@pytest.mark.asyncio
async def test_scenario3_calculation():
"""
Scenario 3: Mathematical Calculation
Expected: Uses calculate tool for accurate results
Performance target: < 15s
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"Calculate 2847 * 1923 + 5612 - 999"
)
tester.assert_response_quality(
result,
min_length=20,
max_time=15.0
)
assert result["success"], "Request failed"
# Verify correct answer: 5,479,394
response_clean = result["response"].replace(",", "").replace(" ", "")
assert "5479394" in response_clean, \
"Calculation result not found in response"
print(f"✓ Calculation query: {result['total_time']:.2f}s")
@pytest.mark.asyncio
async def test_scenario4_date_operations():
"""
Scenario 4: Date/Time Operations
Expected: Uses date/time tools for accurate results
Performance target: < 15s
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"What is the current date and what will the date be 30 days from now?"
)
tester.assert_response_quality(
result,
expected_keywords=["date", "2025"],
min_length=50,
max_time=15.0
)
assert result["success"], "Request failed"
print(f"✓ Date operation query: {result['total_time']:.2f}s")
@pytest.mark.asyncio
async def test_scenario5_multi_tool_reasoning():
"""
Scenario 5: Multi-Tool Complex Query
Expected: Multiple tool calls, synthesizes results
Performance target: < 30s
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"Get the current time in New York and Tokyo, then calculate the time difference in hours"
)
tester.assert_response_quality(
result,
expected_keywords=["time", "hour"],
min_length=80,
max_time=30.0
)
assert result["success"], "Request failed"
print(f"✓ Multi-tool query: {result['total_time']:.2f}s")
@pytest.mark.asyncio
async def test_scenario6_dns_lookup():
"""
Scenario 6: DNS Lookup
Expected: Uses DNS lookup tool from core-api (discovered via OpenAPI)
Performance target: < 15s
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"What are the A records for github.com? Use the core-api DNS lookup tool."
)
tester.assert_response_quality(
result,
expected_keywords=["github", "record"],
min_length=30,
max_time=15.0
)
assert result["success"], "Request failed"
print(f"✓ DNS lookup query: {result['total_time']:.2f}s")
@pytest.mark.asyncio
async def test_tools_disabled():
"""
Test: Agent with Tools Disabled
Expected: Works without tool access, generates knowledge-based response
"""
async with AIFlowTester() as tester:
result = await tester.chat(
"What is Python?",
enable_tools=False
)
tester.assert_response_quality(
result,
expected_keywords=["python", "programming"],
min_length=30,
max_time=10.0
)
assert result["tools_count"] == 0, "Tools should be disabled"
print(f"✓ No-tools query: {result['total_time']:.2f}s")
# ============================================================================
# PERFORMANCE BENCHMARKS
# ============================================================================
@pytest.mark.asyncio
async def test_performance_baseline():
"""
Performance Baseline Test
Establishes baseline metrics for comparison across versions
"""
async with AIFlowTester() as tester:
queries = [
"What is containerization?",
"Explain Kubernetes in one sentence",
"What does API stand for?",
]
times = []
for query in queries:
result = await tester.chat(query)
assert result["success"], f"Query failed: {query}"
times.append(result["total_time"])
avg_time = sum(times) / len(times)
print(f"\n✓ Performance baseline:")
print(f" Average response time: {avg_time:.2f}s")
print(f" Range: {min(times):.2f}s - {max(times):.2f}s")
# Assert reasonable performance
assert avg_time < 10.0, f"Average response too slow: {avg_time:.2f}s"
# ============================================================================
# REGRESSION TESTS
# ============================================================================
@pytest.mark.asyncio
async def test_regression_tool_availability():
"""
Regression: Verify all expected tools are available
"""
async with AIFlowTester() as tester:
response = await tester.client.get(f"{BASE_URL}/v1/tools")
response.raise_for_status()
data = response.json()
tools = {tool["name"] for tool in data["tools"]}
# Expected tools from local.py
expected_tools = {
"get_current_time",
"get_current_date",
"calculate_date_difference",
"add_days_to_date",
"calculate",
"web_search"
}
for tool in expected_tools:
assert tool in tools, f"Missing tool: {tool}"
print(f"✓ Tool availability: {len(tools)} tools registered")
@pytest.mark.asyncio
async def test_regression_health_check():
"""
Regression: Health check endpoint works
"""
async with AIFlowTester() as tester:
response = await tester.client.get(f"{BASE_URL}/health")
response.raise_for_status()
data = response.json()
assert data["status"] == "ok", "Service not healthy"
assert data["service"] == "core-ai"
assert data["agents"]["ollama-native"] is True, \
"OllamaNativeAgent not available"
print(f"✓ Health check passed")
print(f" Default agent: {data['default_agent']}")
print(f" Tools count: {data['tools_count']}")
# ============================================================================
# MAIN RUNNER (for standalone execution)
# ============================================================================
async def run_full_quality_check():
"""Run all tests and generate comprehensive report"""
tester = AIFlowTester()
test_scenarios = [
{
"name": "Scenario 1: Simple Knowledge",
"query": "What is Docker?",
"test": test_scenario1_simple_knowledge
},
{
"name": "Scenario 2: Web Search",
"query": "What are the latest Python 3.13 features?",
"test": test_scenario2_web_search
},
{
"name": "Scenario 3: Calculation",
"query": "Calculate 2847 * 1923 + 5612 - 999",
"test": test_scenario3_calculation
},
{
"name": "Scenario 4: Date Operations",
"query": "What is the current date and what will it be in 30 days?",
"test": test_scenario4_date_operations
},
{
"name": "Scenario 5: Multi-tool Reasoning",
"query": "Get current time in NYC and Tokyo, calculate difference",
"test": test_scenario5_multi_tool_reasoning
},
{
"name": "Scenario 6: DNS Lookup",
"query": "What are the A records for github.com? Use the core-api DNS lookup tool.",
"test": test_scenario6_dns_lookup
},
]
print("\n" + "=" * 80)
print("CORE-AI QUALITY CHECK")
print(f"Started: {datetime.now().isoformat()}")
print(f"Git Tag: {tester.git_info['tag']}")
print(f"Git Commit: {tester.git_info['commit']}")
print("=" * 80 + "\n")
async with tester:
for scenario in test_scenarios:
print(f"\nRunning: {scenario['name']}")
print(f"Query: {scenario['query']}")
print("-" * 80)
try:
await scenario["test"]()
tester.results.append({
"name": scenario["name"],
"query": scenario["query"],
"result": {"success": True, "total_time": 0, "tools_count": 0, "response": ""},
"passed": True
})
except Exception as e:
tester.results.append({
"name": scenario["name"],
"query": scenario["query"],
"result": {"success": False, "error": str(e), "total_time": 0, "tools_count": 0, "response": ""},
"passed": False,
"error": str(e)
})
print(f"✗ FAILED: {e}")
# Generate and print report
report = tester.generate_report()
print("\n" + report)
# Ensure reports directory exists
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
# Save report to file
timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
report_file = REPORTS_DIR / f"quality-report-{timestamp}.txt"
with open(report_file, "w") as f:
f.write(report)
print(f"\nReport saved to: {report_file}")
# Also save as JSON for programmatic analysis
json_file = REPORTS_DIR / f"quality-report-{timestamp}.json"
json_data = {
"timestamp": datetime.now().isoformat(),
"git_info": tester.git_info,
"summary": {
"total": len(tester.results),
"passed": sum(1 for r in tester.results if r.get("passed", False)),
"failed": sum(1 for r in tester.results if not r.get("passed", False))
},
"results": tester.results
}
with open(json_file, "w") as f:
json.dump(json_data, f, indent=2)
print(f"JSON report saved to: {json_file}")
if __name__ == "__main__":
asyncio.run(run_full_quality_check())