github/awesome-kubernetes
1
0
Fork 0
mirror of https://github.com/nubenetes/awesome-kubernetes.git synced 2026-07-28 01:21:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement multi-agent Analyst-Auditor workflow for robust V2 tagging

Browse Source
This commit is contained in:
Nubenetes Bot
2026-05-19 20:57:24 +02:00
parent 0296ca587a
commit 29892a9848
4 changed files with 157 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
GEMINI.md
View File

@@ -126,7 +126,7 @@ This file contains the accumulated instructions and long-term vision for the aut
- **Star Consistency**: Maintain the 1-5 star scale for technical impact. Resources with 0 stars are considered "Standard References" and do not display a star prefix/suffix in the V2 UI.
38. **V2 Semantic Connectivity**: All V2 content generation MUST implement the **Semantic Cross-Linking Engine**. AI agents must autonomously identify related architectural patterns within the same strategic dimension and inject "💡 Explore Related" navigation blocks at the end of sections to facilitate a connected knowledge graph.
39. **Industrial Learning Flow**: V2 documents MUST follow an O'Reilly-style technical progression. Organization within sections must move from foundational theory and standards to advanced implementation details and emerging patterns.
40. **Autonomous Multi-Tagging**: Every resource evaluation MUST assign at least one maturity tag and one resource-type tag if applicable. Fallback to `[COMMUNITY-TOOL]` is only permitted after exhaustive classification failure.
40. **Robust AI-Driven Multi-Tagging**: Every resource evaluation MUST utilize **Real-time Web Grounding (MCP)** to assign 1 to 3 maturity/type tags from the official taxonomy. AI-assigned tags take precedence over static rules. Fallback to `[COMMUNITY-TOOL]` is only permitted after exhaustive classification failure or when no clear maturity signal is found.
- **V2 Index Branding Protection**: The header and vision block of the V2 Elite Portal MUST NOT be modified. The title MUST remain "Nubenetes Elite Portal (V2) | Awesome Kubernetes & Cloud [![Awesome](https://cdn.jsdelivr.net/gh/sindresorhus/awesome@d7305f38d29fed78fa85652e3a63e154dd8e8829/media/badge.svg)](https://github.com/sindresorhus/awesome)" and the abstract MUST use the "The High-Density Vision" text as hardcoded in the optimizer logic to maintain industrial-grade branding.
- **V2 Index Visual Standard (Automotive Roots)**: The Nubenetes V2 Elite Portal index MUST feature a centered banner image linked to `kubernetes.io`, followed by the Horatio Nelson Jackson quote and the specific automotive container metaphor image (`images/container_with_cars_v2.png`). This image is a manually provided asset and MUST NOT be regenerated by AI to ensure the preservation of the project's established visual identity.
- **V2 Index Footer Standard**: The V2 index MUST always conclude with the **Maturity Taxonomy** and **Technical Impact** explanation tables. These sections define the industrial-grade classification and visual code (Highlighting/Bold) used throughout the Elite portal and must be preserved across all automated regenerations.

8
src/agentic_curator.py
View File

@@ -121,13 +121,15 @@ async def evaluate_extracted_assets(raw_assets: List[Dict]) -> Dict[str, Dict]:
"- If the community (Reddit, Hacker News) reports the tool as 'unstable', 'abandoned', or 'vaporware', set reputation_penalty: true.\n"
"PHASE 2: LINGUISTIC DIVERSITY & CLASSIFICATION\n"
"- Identify TECHNICAL_HIERARCHY: List (max 10 strings) Area > Topic > Subtopics.\n"
"Respond ONLY JSON list: [{\"url\": \"...\", \"impact_score\": int, \"reputation_penalty\": bool, \"reputation_summary\": \"...\", \"pub_date\": \"YYYY-MM-DD\", \"primary_category\": \"...\", \"title\": \"...\", \"desc\": \"...\", \"en_summary\": \"...\", \"language\": \"...\", \"type\": \"...\", \"level\": \"...\", \"technical_hierarchy\": [...], \"is_microservice\": bool}, ...]\n\n"
"PHASE 3: MULTI-DIMENSIONAL TAGGING\n"
"- Assign 1 to 3 tags from: [DE FACTO STANDARD], [ENTERPRISE-STABLE], [EMERGING], [GUIDE], [CASE STUDY], [COMMUNITY-TOOL], [LEGACY].\n"
"Respond ONLY JSON list: [{\"url\": \"...\", \"impact_score\": int, \"reputation_penalty\": bool, \"reputation_summary\": \"...\", \"pub_date\": \"YYYY-MM-DD\", \"primary_category\": \"...\", \"title\": \"...\", \"desc\": \"...\", \"en_summary\": \"...\", \"language\": \"...\", \"type\": \"...\", \"level\": \"...\", \"technical_hierarchy\": [...], \"tags\": [...], \"is_microservice\": bool}, ...]\n\n"
"RESOURCES:\n" + "\n".join([f"- {d['asset']['url']}: (MVQ Penalty: {d['mvq_penalty']}) {d['content']}" for d in batch_data])
)
try:
# ENABLE GROUNDING FOR REPUTATION FILTER
results = await call_gemini_with_retry(prompt, use_grounding=True)
results = await call_gemini_with_retry(prompt, use_grounding=True, role="Curator")
if isinstance(results, list):
res_map = {normalize_url(r.get("url", "")): r for r in results}
for d in batch_data:
@@ -145,7 +147,7 @@ async def evaluate_extracted_assets(raw_assets: List[Dict]) -> Dict[str, Dict]:
"title": data["title"], "description": data["desc"], "ai_summary": data.get("en_summary", data["desc"]),
"language": data.get("language", "English"), "resource_type": data.get("type", "Reference"),
"complexity": data.get("level", "Intermediate"), "hierarchy": data.get("technical_hierarchy", ["General"]),
"is_microservice": data.get("is_microservice", False), "year": data.get("pub_date", "N/A")[:4],
"tags": data.get("tags", []), "is_microservice": data.get("is_microservice", False), "year": data.get("pub_date", "N/A")[:4],
"stars": min(max(score // 20, 0), 5), "content_hash": d["hash"],
"reputation_status": "Vetted" if not data.get("reputation_penalty") else "Suspicious",
"reputation_summary": data.get("reputation_summary", ""),

28
src/gemini_utils.py
View File

@@ -33,9 +33,13 @@ class GeminiSessionTracker:
self.cache_hits += 1
self.est_tokens_saved += est_tokens
def track_call(self, key_idx: int, model: str, status: int, usage: Dict = None):
def track_call(self, key_idx: int, model: str, status: int, usage: Dict = None, role: str = "General"):
if status == 200:
self.model_usage[model] = self.model_usage.get(model, 0) + 1
# Track by Role for Agentic Observability
self.key_stats[key_idx].setdefault("roles", {})
self.key_stats[key_idx]["roles"][role] = self.key_stats[key_idx]["roles"].get(role, 0) + 1
if usage:
self.total_tokens_prompt += usage.get("promptTokenCount", 0)
self.total_tokens_completion += usage.get("candidatesTokenCount", 0)
@@ -48,13 +52,23 @@ class GeminiSessionTracker:
def get_intelligence_report(self) -> str:
report = "\n### 🧠 AI Intelligence & Observability Report\n\n"
report += "#### 🤖 Model Selection Logic (Dynamic)\n"
report += f"Execution started with discovery of top models based on May 2026 hierarchy.\n\n"
report += "#### 🤖 Agentic Roles & Model Selection (Dynamic)\n"
report += f"Execution utilized a multi-agent Analyst-Auditor workflow for maximum robustness.\n\n"
report += "| Agent Role | Model Used | Successes |\n| :--- | :--- | :---: |\n"
role_stats = {}
for idx, stats in self.key_stats.items():
for role, count in stats.get("roles", {}).items():
role_stats[role] = role_stats.get(role, 0) + count
for role, count in role_stats.items():
report += f"| **{role}** | Dynamic Selection | **{count}** |\n"
report += "\n#### 🤖 Model Performance Matrix\n"
report += "| Model Used | Successful Calls | Hierarchy Logic |\n| :--- | :---: | :--- |\n"
usage_items = sorted(self.model_usage.items(), key=lambda x: x[1], reverse=True)
for model, count in usage_items:
logic = "Elite/Pro (Complex Reasoning)" if "pro" in model else "Flash/Lite (High Speed)"
logic = "Elite Auditor (High Reasoning)" if "pro" in model else "Fast Analyst (High Speed)"
report += f"| `{model}` | **{count}** | {logic} |\n"
if not self.model_usage: report += "| No AI calls | 0 | N/A |\n"
@@ -252,7 +266,7 @@ def normalize_url(url: str) -> str:
def is_fuzzy_duplicate(url_a: str, url_b: str) -> bool:
return normalize_url(url_a) == normalize_url(url_b)
async def call_gemini_with_retry(prompt: str, response_format: str = "json", max_retries: int = 3, prefer_flash: bool = False, use_grounding: bool = False):
async def call_gemini_with_retry(prompt: str, response_format: str = "json", max_retries: int = 3, prefer_flash: bool = False, use_grounding: bool = False, role: str = "General"):
global CURRENT_KEY_INDEX, GLOBAL_COOLDOWN_UNTIL
if not GEMINI_API_KEYS: raise ValueError("No GEMINI_API_KEYS configured.")
@@ -297,7 +311,7 @@ async def call_gemini_with_retry(prompt: str, response_format: str = "json", max
except: pass
usage = resp_json.get("usageMetadata", {})
SESSION_TRACKER.track_call(current_idx, model, response.status_code, usage)
SESSION_TRACKER.track_call(current_idx, model, response.status_code, usage, role=role)
if response.status_code == 200:
CURRENT_KEY_INDEX = current_idx
@@ -349,7 +363,7 @@ async def call_gemini_with_retry(prompt: str, response_format: str = "json", max
break
except Exception as e:
SESSION_TRACKER.track_call(current_idx, model, 0, {})
SESSION_TRACKER.track_call(current_idx, model, 0, {}, role=role)
diagnostics.add_attempt(model, 0, str(e))
break

171
src/v2_optimizer.py
View File

@@ -47,6 +47,8 @@ class V2VisionEngine:
"- Order hierarchy to facilitate a structured learning journey.\n"
"PHASE 4: MANDATORY DESCRIPTIONS\n"
"- If 'Current Desc' is empty, generate a professional summary. Style: O'Reilly technical.\n"
"PHASE 5: ADVANCED MATURITY TAGGING\n"
"- Assign 1 to 3 tags from: [DE FACTO STANDARD], [ENTERPRISE-STABLE], [EMERGING], [GUIDE], [CASE STUDY], [COMMUNITY-TOOL], [LEGACY].\n"
)
self.inventory = self._load_inventory()
self.maturity_audit = []
@@ -197,6 +199,7 @@ class V2VisionEngine:
enrich_metadata = os.getenv("ENRICH_METADATA", "false").lower() == "true"
special_files = [sa["file"] for sa in self.special_assets_rules.get("special_assets", [])]
# Mandate 15: Proactive Enrichment for V2 (GitHub metadata is critical for tags)
for l in links:
item = l.copy()
norm_url = normalize_url(l["url"])
@@ -208,15 +211,16 @@ class V2VisionEngine:
match = re.search(r'github\.com/([^/]+/[^/]+)', norm_url)
if match: project_id = match.group(1).lower()
# Mandate 15: If enrichment is ON and gh_metadata is missing, we must fetch it
if enrich_metadata and "github.com" in norm_url:
cached = self.inventory.get(norm_url, {})
if not cached.get("gh_stars"):
log_event(f" [METADATA] Enrichment: Fetching GH Activity for {norm_url}")
# Mandate 43: Always ensure GH metadata for GitHub links in V2 to power [DE FACTO STANDARD] logic
cached = self.inventory.get(norm_url, {})
if "github.com" in norm_url and (enrich_metadata or not cached.get("gh_stars")):
if not cached.get("gh_stars") or enrich_metadata:
log_event(f" [METADATA] V2 Pulse: Fetching GH Activity for {norm_url}")
gh_data = await get_github_activity(norm_url)
if gh_data:
if norm_url not in self.inventory: self.inventory[norm_url] = {}
self.inventory[norm_url].update(gh_data)
item.update(gh_data)
if not force_eval and norm_url in self.inventory and "stars" in self.inventory[norm_url]:
cached = self.inventory[norm_url]
@@ -238,67 +242,152 @@ class V2VisionEngine:
to_evaluate.append(item)
if to_evaluate:
for i in range(0, len(to_evaluate), 50):
batch = to_evaluate[i:i+50]
prompt = (f"{self.library_criteria}\nRespond ONLY JSON: {{\"results\": [{{ \"idx\": int, \"year\": \"YYYY\", \"stars\": 0-5, \"hierarchy\": [\"Area\", \"Topic\", ...], \"summary\": \"...\", \"language\": \"...\", \"type\": \"...\", \"complexity\": \"...\", \"is_microservice\": bool }}, ...]}}\n\nLINKS:\n" +
"\n".join([f"{idx}. {l['title']} ({l['url']})" for idx, l in enumerate(batch)]))
# Mandate 32 & 40: Smaller batches (10) for high-precision tagging and grounding efficiency
BATCH_SIZE = 10
from src.mandate_ingestor import get_system_mandates
dynamic_mandates = get_system_mandates()
log_event(f"[*] Agent Phase 1: Analyst Evaluation ({len(to_evaluate)} resources)...")
analyst_results = []
for i in range(0, len(to_evaluate), BATCH_SIZE):
batch = to_evaluate[i:i+BATCH_SIZE]
# Analyst Prompt (Focus: Classification & Summary)
prompt = (
f"You are the Nubenetes Technical Analyst (2026).\n"
f"{dynamic_mandates}\n"
f"{self.library_criteria}\n"
"PHASE 5: INITIAL TAGGING\n"
"- Assign 1 to 3 preliminary tags.\n"
"Respond ONLY JSON: {{\"results\": [{{ \"idx\": int, \"year\": \"YYYY\", \"stars\": 0-5, \"hierarchy\": [\"Area\", \"Topic\", ...], \"tags\": [\"...\"], \"summary\": \"...\", \"language\": \"...\", \"type\": \"...\", \"complexity\": \"...\", \"is_microservice\": bool }}, ...]}}\n\n"
"LINKS:\n" + "\n".join([f"{idx}. {l['title']} ({l['url']})" for idx, l in enumerate(batch)])
)
try:
# ENABLE GROUNDING FOR V2 (High-Density Accuracy)
data = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True)
data = await call_gemini_with_retry(prompt, prefer_flash=True, use_grounding=True, role="Analyst")
for res in data.get("results", []):
idx = int(res["idx"])
if idx < len(batch):
item = batch[idx].copy()
norm_url = normalize_url(item["url"])
p_id = norm_url
if "github.com" in norm_url:
m = re.search(r'github\.com/([^/]+/[^/]+)', norm_url)
if m: p_id = m.group(1).lower()
eval_data = {
"year": str(res.get("year", "N/A")), "stars": min(max(int(res.get("stars", 0)), 0), 5),
"ai_summary": res.get("summary", ""), "language": res.get("language", "English"),
"resource_type": res.get("type", "Reference"), "complexity": res.get("complexity", "Intermediate"),
"hierarchy": res.get("hierarchy", ["General"]), "is_microservice": bool(res.get("is_microservice", False)),
"hierarchy": res.get("hierarchy", ["General"]), "tags": res.get("tags", []),
"is_microservice": bool(res.get("is_microservice", False)),
"status": "online", "is_special": item.get("is_special", False)
}
item.update(eval_data)
self.inventory[norm_url] = eval_data
self.inventory[norm_url]["title"] = item["title"]
if p_id not in project_registry or item["stars"] > project_registry[p_id].get("stars", 0):
if p_id in project_registry and project_registry[p_id].get("is_special"): item["is_special"] = True
project_registry[p_id] = item
analyst_results.append(item)
except:
for l in batch:
u = normalize_url(l["url"])
if u not in project_registry: project_registry[u] = l
for l in batch: analyst_results.append(l)
await asyncio.sleep(0.3)
# --- AGENT PHASE 2: SELECTIVE AUDIT (MCP-Grounded) ---
# Identify candidates for high-trust verification
audit_candidates = [l for l in analyst_results if "[DE FACTO STANDARD]" in l.get("tags", []) or "[ENTERPRISE-STABLE]" in l.get("tags", [])]
if audit_candidates:
log_event(f"[*] Agent Phase 2: Auditor Verification ({len(audit_candidates)} high-impact candidates)...")
# AUDIT BATCH: Very small for max grounding precision
for i in range(0, len(audit_candidates), 5):
batch = audit_candidates[i:i+5]
audit_prompt = (
f"You are the Nubenetes Auditor (2026).\n"
f"{dynamic_mandates}\n"
"MISSION: Verify the 'Elite' status of these resources using your GOOGLE_SEARCH tool.\n"
"CRITERIA:\n"
"- [DE FACTO STANDARD]: Industry baseline, used by everyone (e.g. K8s, Terraform).\n"
"- [ENTERPRISE-STABLE]: Proven, high-trust, supported (e.g. ArgoCD, Linkerd).\n"
"- REPUTATION: Check Reddit/Hacker News for stability or abandonment reports.\n"
"Respond ONLY JSON: {{\"audits\": [{{ \"idx\": int, \"verified_tags\": [\"...\"], \"reputation_summary\": \"...\", \"reputation_penalty\": bool }}, ...]}}\n\n"
"RESOURCES TO AUDIT:\n" + "\n".join([f"{idx}. {l['title']} ({l['url']}) - Proposed: {l.get('tags')}" for idx, l in enumerate(batch)])
)
try:
# AUDIT USES PRO MODEL (High Reasoning) + GROUNDING (Live Data)
audit_data = await call_gemini_with_retry(audit_prompt, prefer_flash=False, use_grounding=True, role="Auditor")
for aud in audit_data.get("audits", []):
idx = int(aud["idx"])
if idx < len(batch):
url = batch[idx]["url"]
# Update tags and add reputation metadata (Mandate 32/33)
batch[idx]["tags"] = aud.get("verified_tags", batch[idx]["tags"])
batch[idx]["reputation_summary"] = aud.get("reputation_summary", "")
if aud.get("reputation_penalty"):
batch[idx]["stars"] = max(batch[idx].get("stars", 1) - 1, 1)
if "[DE FACTO STANDARD]" in batch[idx]["tags"]: batch[idx]["tags"].remove("[DE FACTO STANDARD]")
except: pass
await asyncio.sleep(0.5)
# Finalize Registry
for item in analyst_results:
norm_url = normalize_url(item["url"])
p_id = norm_url
if "github.com" in norm_url:
m = re.search(r'github\.com/([^/]+/[^/]+)', norm_url)
if m: p_id = m.group(1).lower()
# Persist to inventory
self.inventory[norm_url] = {k:v for k,v in item.items() if k not in ["url", "title", "original_file", "is_special", "aliases"]}
self.inventory[norm_url]["title"] = item["title"]
if p_id not in project_registry or item.get("stars", 0) > project_registry[p_id].get("stars", 0):
if p_id in project_registry and project_registry[p_id].get("is_special"): item["is_special"] = True
project_registry[p_id] = item
return list(project_registry.values())
def _calculate_tags(self, item: Dict) -> List[str]:
tags = []
"""
Mandate 40: Multi-Dimensional Tagging (1:N).
Merges AI-assigned tags with rule-based maturity signals to ensure high-fidelity classification.
Utilizes MCP-style grounding data (GitHub stars, resource types) to override generic defaults.
"""
# 0. Collect all possible tag sources
ai_tags = item.get("tags", [])
if isinstance(ai_tags, str): ai_tags = [ai_tags] # Resiliency
valid_set = {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]", "[EMERGING]", "[GUIDE]", "[CASE STUDY]", "[COMMUNITY-TOOL]", "[LEGACY]"}
# Start with filtered AI tags
tags = set([t for t in ai_tags if t in valid_set])
# 1. GitHub Objective Reality (Mandate 43)
raw_gh = item.get("gh_stars", 0)
gh_stars = int(raw_gh) if str(raw_gh).isdigit() else 0
curator_stars = item.get("stars", 0)
res_type = item.get("resource_type", "Reference").lower()
# 1. Maturity Logic (GitHub based OR curator based)
curator_stars = int(item.get("stars", 0))
if gh_stars > 15000 or curator_stars >= 5:
tags.append("[DE FACTO STANDARD]")
tags.add("[DE FACTO STANDARD]")
if "[COMMUNITY-TOOL]" in tags: tags.remove("[COMMUNITY-TOOL]")
elif gh_stars > 3000 or curator_stars >= 4:
tags.append("[ENTERPRISE-STABLE]")
tags.add("[ENTERPRISE-STABLE]")
if "[COMMUNITY-TOOL]" in tags: tags.remove("[COMMUNITY-TOOL]")
# 2. Type Mapping (AI based)
if "guide" in res_type or "tutorial" in res_type: tags.append("[GUIDE]")
if "case study" in res_type or "report" in res_type: tags.append("[CASE STUDY]")
# 2. Type Mapping (AI based labels)
res_type = item.get("resource_type", "Reference").lower()
if any(x in res_type for x in ["guide", "tutorial", "hands-on", "learning", "course"]):
tags.add("[GUIDE]")
if any(x in res_type for x in ["case study", "report", "whitepaper", "success story", "usage"]):
tags.add("[CASE STUDY]")
# 3. Emerging / Legacy logic
if item.get("complexity") == "Cutting Edge" or "emerging" in item.get("ai_summary", "").lower():
tags.append("[EMERGING]")
ai_summary = item.get("ai_summary", "").lower()
complexity = item.get("complexity", "Intermediate")
if complexity == "Cutting Edge" or "emerging" in ai_summary or "experimental" in ai_summary or "alpha" in ai_summary:
tags.add("[EMERGING]")
if "legacy" in ai_summary or "deprecated" in ai_summary or "archived" in ai_summary or "v1-only" in ai_summary:
tags.add("[LEGACY]")
# 4. Fallback: Only use [COMMUNITY-TOOL] if no other maturity tag is present
maturity_tags = {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]", "[EMERGING]", "[LEGACY]"}
if not (tags & maturity_tags):
tags.add("[COMMUNITY-TOOL]")
# Fallback
if not tags: tags.append("[COMMUNITY-TOOL]")
return tags
# Clean up: If we have high maturity, remove community-tool
if (tags & {"[DE FACTO STANDARD]", "[ENTERPRISE-STABLE]"}) and "[COMMUNITY-TOOL]" in tags:
tags.remove("[COMMUNITY-TOOL]")
return sorted(list(tags))
async def _rebuild_structure(self, library_inventory: List[Dict]):
special_rules = {sa["file"]: sa for sa in self.special_assets_rules.get("special_assets", [])}