Mayne0213/mas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

REFACTOR(app): use Claude API only

Browse Source 
- Remove Groq API integration
- Use only Anthropic Claude API
This commit is contained in:
Mayne0213
2025-12-24 18:29:35 +09:00
parent a382985c79
commit 48012c4864
13 changed files with 334 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
README.md
View File

@@ -1,21 +1,39 @@
# MAS (Multi-Agent System)
MAS is a unified UI and orchestration layer for multiple AI agents (similar to ChatGPT, Claude, Gemini), running on your own Kubernetes cluster.
**K8s Infrastructure Planning System** - AI agents that analyze your Kubernetes cluster and generate implementation plans.
## 🎯 Architecture
## 🎯 What is this?
### Agents
- **Claude Code (Orchestrator)**: overall coordinator & DevOps expert
- **Qwen Backend**: backend engineer (FastAPI, Node.js)
- **Qwen Frontend**: frontend engineer (Next.js, React)
- **Qwen SRE**: monitoring & reliability engineer
MAS는 Kubernetes 클러스터 상태를 분석하고, 인프라 배포 계획을 수립하는 AI 에이전트 시스템입니다.
**사용 시나리오:**
1. "Tekton을 도입하고 싶어" → 클러스터 분석 → YAML 구조 설계 → 구현 가이드 생성
2. 생성된 Markdown 프롬프트를 복사해서 다른 AI (Claude Code, ChatGPT 등)에 붙여넣기
3. 실제 코드 구현은 다른 AI가 담당
## 🤖 Agents
### Planning Agent (Claude 4.5)
- 폴더 구조 설계 (deploy/tool/base, overlays/prod, etc.)
- YAML 파일 조직화
- K8s 리소스 계획 (Namespace, Deployment, Service, etc.)
### Research Agent (Groq Llama 3.3)
- kubectl 명령어로 클러스터 상태 분석
- 기존 리소스 확인 (namespaces, storage classes, quotas)
- 호환성 검토
### Prompt Generator (Claude 4.5)
- Markdown 형식의 구현 가이드 생성
- YAML 예시 포함
- 검증 명령어 제공
### Tech stack
- **Backend**: LangGraph + LangChain + FastAPI
- **UI**: Chainlit (chat-style UI)
- **Database**: PostgreSQL (CNPG)
- **Cache**: Redis
- **LLMs**: Claude API + **Groq Llama 3.x** (OpenAI-compatible API)
- **Backend**: LangGraph + LangChain + FastAPI
- **UI**: Chainlit (chat-style UI)
- **Database**: PostgreSQL (CNPG)
- **Cache**: Redis
- **LLMs**: Claude API (Orchestrator, Planning, Prompt Gen) + Groq Llama 3.3 (Research)
- **Deploy**: Kubernetes + ArgoCD
---
@@ -195,61 +213,82 @@ Examples:
## 📝 Usage examples
### Backend API request
### Example 1: Deploy Tekton
```text
User: "Create a signup API with FastAPI.
Use PostgreSQL and JWT tokens."
User: "Tekton을 도입하고 싶어"
🎼 Orchestrator:
→ routes to Qwen Backend
→ routes to Planning Agent
⚙️ Qwen Backend:
→ generates FastAPI router, Pydantic models, DB schema, JWT logic
📋 Planning Agent:
→ designs folder structure: deploy/tekton/{base,overlays/prod}
→ plans K8s resources: Namespace, RBAC, Deployments, Services
→ identifies research needs
🎼 Orchestrator:
→ reviews, suggests improvements, and outputs final code snippet & file layout
🔍 Research Agent:
→ runs: kubectl get namespaces, kubectl get storageclasses
→ checks: existing tekton resources, cluster version
→ analyzes: available resources and quotas
📝 Prompt Generator:
→ generates comprehensive Markdown implementation guide
→ includes: YAML examples, folder structure, validation commands
✨ Output: Markdown prompt ready to copy-paste into Claude Code/ChatGPT
```
### Frontend component request
### Example 2: Deploy Harbor Registry
```text
User: "Build a responsive dashboard chart component using Recharts."
User: "Harbor를 배포하려고 해"
🎼 Orchestrator:
→ routes to Qwen Frontend
→ Planning: folder structure + YAML organization
→ Research: storage classes, ingress controllers, TLS setup
→ Prompt Gen: Markdown guide with Harbor Helm values, ingress config, etc.
🎨 Qwen Frontend:
→ generates a Next.js/React component with TypeScript and responsive styles
🎼 Orchestrator:
→ explains how to integrate it into your existing app
✨ Copy the prompt → Paste into another AI → Get actual implementation
```
### Infra / SRE request
### Example 3: Deploy Prometheus
```text
User: "Prometheus is firing high memory alerts for the PostgreSQL pod.
Help me stabilize it."
User: "Prometheus를 설치하고 싶어"
🎼 Orchestrator:
→ routes to Qwen SRE
→ Planning: monitoring stack structure (Prometheus, Grafana, AlertManager)
→ Research: existing ServiceMonitors, PVC requirements
→ Prompt Gen: Complete implementation guide
📊 Qwen SRE:
→ analyzes metrics & logs (conceptually),
proposes tuning (Postgres config, indexes, pooler),
and suggests alert threshold adjustments.
✨ Result: Ready-to-use prompt for code generation
```
---
## 🔧 Workflow
```
User Input: "Deploy X"
↓
Orchestrator (조율)
↓
Planning Agent (구조 설계)
↓
Research Agent (클러스터 분석)
↓
Prompt Generator (가이드 생성)
↓
Output: Markdown Implementation Guide
↓
User copies → Pastes to Claude Code/ChatGPT → Gets actual code
```
## 🤝 Contributing
Contributions are welcome:
- New agents (e.g., data engineer, security engineer)
- New tools (Harbor, Tekton, CNPG, MetalLB integrations)
- Better prompts and workflows
- Docs and examples
- Improve Planning Agent prompts for better folder structures
- Enhance Research Agent kubectl commands
- Add more infrastructure tools (Harbor, Tekton, CNPG, MetalLB, etc.)
- Better Markdown template for Prompt Generator
Feel free to open issues or PRs in your Git repository.

2
deploy/k8s/overlays/prod/kustomization.yaml
View File

@@ -14,5 +14,5 @@ commonLabels:
# 이미지 태그 설정 (ArgoCD Image Updater가 자동으로 업데이트)
images:
- name: gitea0213.kro.kr/bluemayne/mas
newTag: main-sha-422bf3e092c3f7c1228fad9fba7f8d615b8bcdba
newTag: main-sha-bcd28d7d2120843e827ee23014bc94b0b418d565

11
services/backend/agents/__init__.py
View File

@@ -1,22 +1,17 @@
"""
MAS Agents Package
K8s Infrastructure Planning System
"""
from .state import AgentState
from .orchestrator import orchestrator_node
from .planning_agent import planning_node
from .research_agent import research_node
from .code_backend_agent import backend_code_node
from .code_frontend_agent import frontend_code_node
from .code_infrastructure_agent import infrastructure_code_node
from .review_agent import review_node
from .prompt_generator_agent import prompt_generator_node
__all__ = [
'AgentState',
'orchestrator_node',
'planning_node',
'research_node',
'backend_code_node',
'frontend_code_node',
'infrastructure_code_node',
'review_node',
'prompt_generator_node',
]

126
services/backend/agents/code_backend_agent.py
View File

@@ -1,126 +0,0 @@
"""
Backend Code Agent (Groq)
백엔드 코드 작성/수정 전문 (FastAPI, Node.js, Database)
"""
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from .state import AgentState
from tools.bash_tool import bash_tools
import os
# Groq 모델 초기화
groq_backend = ChatOpenAI(
model="llama-3.3-70b-versatile",
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1",
temperature=0.5
)
BACKEND_PROMPT = """You are the Backend Code Agent.
## Role
Write backend code (FastAPI, Node.js, databases).
## Tools
- execute_host: Write files to /home/ubuntu/Projects/, run git commands
- execute_bash: Test and validate
## Important
- After modifying files: git add, commit, and push (ArgoCD deploys automatically)
- Write clean, secure code with proper error handling
"""
def backend_code_node(state: AgentState) -> AgentState:
"""
Backend Code 노드: 백엔드 코드 작성
"""
messages = state["messages"]
task_plan = state.get("task_plan", {})
research_data = state.get("research_data", {})
# Groq에 bash 도구 바인딩
groq_with_tools = groq_backend.bind_tools(bash_tools)
# 코드 작성 요청 구성
code_request = f"""
작업 계획: {task_plan.get('summary', '')}
수집된 정보: {research_data.get('summary', '')}
다음 백엔드 코드를 작성해주세요.
"""
# Groq 호출
response = groq_with_tools.invoke([
SystemMessage(content=BACKEND_PROMPT),
HumanMessage(content=code_request)
])
# Tool calls 처리
tool_outputs = []
if hasattr(response, 'tool_calls') and response.tool_calls:
for tool_call in response.tool_calls:
try:
tool_name = tool_call.get('name') or tool_call.get('function', {}).get('name', 'unknown')
tool_args_raw = tool_call.get('args') or tool_call.get('function', {}).get('arguments', {})
# tool_args가 문자열인 경우 JSON 파싱 시도
import json
if isinstance(tool_args_raw, str):
try:
tool_args = json.loads(tool_args_raw)
except json.JSONDecodeError:
# JSON 파싱 실패 시 빈 딕셔너리 사용
tool_args = {}
print(f"⚠️ Failed to parse tool_args as JSON: {tool_args_raw}")
elif isinstance(tool_args_raw, dict):
tool_args = tool_args_raw
else:
tool_args = {}
print(f"⚠️ Unexpected tool_args type: {type(tool_args_raw)}")
# tool_name에 따라 올바른 도구 선택
from tools.bash_tool import execute_bash, execute_host
if tool_name == "execute_host":
tool_func = execute_host
else:
tool_func = execute_bash
# 필수 파라미터 확인
if 'command' not in tool_args:
tool_outputs.append(f"\n❌ **{tool_name}** failed: 'command' parameter is required")
continue
tool_result = tool_func.invoke(tool_args)
tool_outputs.append(f"\n🔧 **{tool_name}({tool_args.get('command', '')[:50]}...)**:\n{tool_result}")
except Exception as e:
error_detail = str(e)
import traceback
print(f"❌ Tool call error: {error_detail}")
print(traceback.format_exc())
tool_outputs.append(f"\n❌ **{tool_name if 'tool_name' in locals() else 'unknown'}** failed: {error_detail}")
# Tool 결과와 함께 재호출
if tool_outputs:
tool_context = "\n".join(tool_outputs)
response = groq_backend.invoke([
SystemMessage(content=BACKEND_PROMPT),
HumanMessage(content=code_request),
HumanMessage(content=f"도구 실행 결과:\n{tool_context}\n\n작업 결과를 요약해주세요.")
])
content = response.content
if tool_outputs:
content = "\n".join(tool_outputs) + "\n\n" + content
# 상태 업데이트
state["code_outputs"]["backend"] = content
state["messages"].append({
"role": "backend_developer",
"content": content
})
state["current_agent"] = "orchestrator"
return state

126
services/backend/agents/code_frontend_agent.py
View File

@@ -1,126 +0,0 @@
"""
Frontend Code Agent (Groq)
프론트엔드 코드 작성/수정 전문 (React, Next.js, Vue)
"""
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from .state import AgentState
from tools.bash_tool import bash_tools
import os
# Groq 모델 초기화
groq_frontend = ChatOpenAI(
model="llama-3.3-70b-versatile",
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1",
temperature=0.5
)
FRONTEND_PROMPT = """You are the Frontend Code Agent.
## Role
Write frontend code (React, Next.js, Vue).
## Tools
- execute_host: Write files to /home/ubuntu/Projects/, run git commands
- execute_bash: Test and validate
## Important
- After modifying files: git add, commit, and push (ArgoCD deploys automatically)
- Write TypeScript, responsive UI, accessible components
"""
def frontend_code_node(state: AgentState) -> AgentState:
"""
Frontend Code 노드: 프론트엔드 코드 작성
"""
messages = state["messages"]
task_plan = state.get("task_plan", {})
research_data = state.get("research_data", {})
# Groq에 bash 도구 바인딩
groq_with_tools = groq_frontend.bind_tools(bash_tools)
# 코드 작성 요청 구성
code_request = f"""
작업 계획: {task_plan.get('summary', '')}
수집된 정보: {research_data.get('summary', '')}
다음 프론트엔드 코드를 작성해주세요.
"""
# Groq 호출
response = groq_with_tools.invoke([
SystemMessage(content=FRONTEND_PROMPT),
HumanMessage(content=code_request)
])
# Tool calls 처리
tool_outputs = []
if hasattr(response, 'tool_calls') and response.tool_calls:
for tool_call in response.tool_calls:
try:
tool_name = tool_call.get('name') or tool_call.get('function', {}).get('name', 'unknown')
tool_args_raw = tool_call.get('args') or tool_call.get('function', {}).get('arguments', {})
# tool_args가 문자열인 경우 JSON 파싱 시도
import json
if isinstance(tool_args_raw, str):
try:
tool_args = json.loads(tool_args_raw)
except json.JSONDecodeError:
# JSON 파싱 실패 시 빈 딕셔너리 사용
tool_args = {}
print(f"⚠️ Failed to parse tool_args as JSON: {tool_args_raw}")
elif isinstance(tool_args_raw, dict):
tool_args = tool_args_raw
else:
tool_args = {}
print(f"⚠️ Unexpected tool_args type: {type(tool_args_raw)}")
# tool_name에 따라 올바른 도구 선택
from tools.bash_tool import execute_bash, execute_host
if tool_name == "execute_host":
tool_func = execute_host
else:
tool_func = execute_bash
# 필수 파라미터 확인
if 'command' not in tool_args:
tool_outputs.append(f"\n❌ **{tool_name}** failed: 'command' parameter is required")
continue
tool_result = tool_func.invoke(tool_args)
tool_outputs.append(f"\n🔧 **{tool_name}({tool_args.get('command', '')[:50]}...)**:\n{tool_result}")
except Exception as e:
error_detail = str(e)
import traceback
print(f"❌ Tool call error: {error_detail}")
print(traceback.format_exc())
tool_outputs.append(f"\n❌ **{tool_name if 'tool_name' in locals() else 'unknown'}** failed: {error_detail}")
# Tool 결과와 함께 재호출
if tool_outputs:
tool_context = "\n".join(tool_outputs)
response = groq_frontend.invoke([
SystemMessage(content=FRONTEND_PROMPT),
HumanMessage(content=code_request),
HumanMessage(content=f"도구 실행 결과:\n{tool_context}\n\n작업 결과를 요약해주세요.")
])
content = response.content
if tool_outputs:
content = "\n".join(tool_outputs) + "\n\n" + content
# 상태 업데이트
state["code_outputs"]["frontend"] = content
state["messages"].append({
"role": "frontend_developer",
"content": content
})
state["current_agent"] = "orchestrator"
return state

126
services/backend/agents/code_infrastructure_agent.py
View File

@@ -1,126 +0,0 @@
"""
Infrastructure Code Agent (Groq)
인프라/DevOps 코드 작성 전문 (Kubernetes, YAML, Docker)
"""
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from .state import AgentState
from tools.bash_tool import bash_tools
import os
# Groq 모델 초기화
groq_infrastructure = ChatOpenAI(
model="llama-3.3-70b-versatile",
api_key=os.getenv("GROQ_API_KEY"),
base_url="https://api.groq.com/openai/v1",
temperature=0.3 # 인프라는 더 정확하게
)
INFRASTRUCTURE_PROMPT = """You are the Infrastructure Code Agent.
## Role
Write Kubernetes manifests, Docker configs, and infrastructure code.
## Tools
- execute_host: Write YAML files to /home/ubuntu/Projects/, run kubectl and git
- execute_bash: Validate YAML
## Important
- After modifying files: git add, commit, and push (ArgoCD deploys automatically)
- Use proper resource limits, health checks, and security contexts
"""
def infrastructure_code_node(state: AgentState) -> AgentState:
"""
Infrastructure Code 노드: 인프라 코드 작성
"""
messages = state["messages"]
task_plan = state.get("task_plan", {})
research_data = state.get("research_data", {})
# Groq에 bash 도구 바인딩
groq_with_tools = groq_infrastructure.bind_tools(bash_tools)
# 코드 작성 요청 구성
code_request = f"""
작업 계획: {task_plan.get('summary', '')}
수집된 정보: {research_data.get('summary', '')}
다음 인프라 코드/YAML을 작성해주세요.
"""
# Groq 호출
response = groq_with_tools.invoke([
SystemMessage(content=INFRASTRUCTURE_PROMPT),
HumanMessage(content=code_request)
])
# Tool calls 처리
tool_outputs = []
if hasattr(response, 'tool_calls') and response.tool_calls:
for tool_call in response.tool_calls:
try:
tool_name = tool_call.get('name') or tool_call.get('function', {}).get('name', 'unknown')
tool_args_raw = tool_call.get('args') or tool_call.get('function', {}).get('arguments', {})
# tool_args가 문자열인 경우 JSON 파싱 시도
import json
if isinstance(tool_args_raw, str):
try:
tool_args = json.loads(tool_args_raw)
except json.JSONDecodeError:
# JSON 파싱 실패 시 빈 딕셔너리 사용
tool_args = {}
print(f"⚠️ Failed to parse tool_args as JSON: {tool_args_raw}")
elif isinstance(tool_args_raw, dict):
tool_args = tool_args_raw
else:
tool_args = {}
print(f"⚠️ Unexpected tool_args type: {type(tool_args_raw)}")
# tool_name에 따라 올바른 도구 선택
from tools.bash_tool import execute_bash, execute_host
if tool_name == "execute_host":
tool_func = execute_host
else:
tool_func = execute_bash
# 필수 파라미터 확인
if 'command' not in tool_args:
tool_outputs.append(f"\n❌ **{tool_name}** failed: 'command' parameter is required")
continue
tool_result = tool_func.invoke(tool_args)
tool_outputs.append(f"\n🔧 **{tool_name}({tool_args.get('command', '')[:50]}...)**:\n{tool_result}")
except Exception as e:
error_detail = str(e)
import traceback
print(f"❌ Tool call error: {error_detail}")
print(traceback.format_exc())
tool_outputs.append(f"\n❌ **{tool_name if 'tool_name' in locals() else 'unknown'}** failed: {error_detail}")
# Tool 결과와 함께 재호출
if tool_outputs:
tool_context = "\n".join(tool_outputs)
response = groq_infrastructure.invoke([
SystemMessage(content=INFRASTRUCTURE_PROMPT),
HumanMessage(content=code_request),
HumanMessage(content=f"도구 실행 결과:\n{tool_context}\n\n작업 결과를 요약해주세요.")
])
content = response.content
if tool_outputs:
content = "\n".join(tool_outputs) + "\n\n" + content
# 상태 업데이트
state["code_outputs"]["infrastructure"] = content
state["messages"].append({
"role": "infrastructure_engineer",
"content": content
})
state["current_agent"] = "orchestrator"
return state

49
services/backend/agents/orchestrator.py
View File

@@ -17,37 +17,39 @@ claude_orchestrator = ChatAnthropic(
)
ORCHESTRATOR_PROMPT = """You are the Orchestrator of a Multi-Agent System.
ORCHESTRATOR_PROMPT = """You are the Orchestrator of a K8s Infrastructure Planning System.
## Role
Coordinate agents to complete user requests efficiently.
Coordinate agents to analyze K8s cluster and generate implementation plans.
## Available Agents
- planning: Create task plans
- research: Gather information (K8s, files, databases)
- code_backend: Write backend code
- code_frontend: Write frontend code
- code_infrastructure: Write K8s/infrastructure code
- review: Review and validate code
- end: Complete the task
- planning: Design folder structure, YAML organization, K8s resources
- research: Analyze K8s cluster state (kubectl commands, resources, configs)
- prompt_generator: Generate Markdown implementation prompt for other AI assistants
- end: Complete the task and show final prompt
## Workflow
1. Analyze user request
2. Delegate to planning agent (if no plan exists)
3. Coordinate research → code → review cycle
4. Limit iterations to 3 maximum
5. Return results to user
1. User requests infrastructure deployment (e.g., "Deploy Tekton")
2. Delegate to **planning** agent (if no plan exists)
3. Delegate to **research** agent to analyze cluster state
4. Delegate to **prompt_generator** to create implementation prompt
5. End with final Markdown prompt for the user
## Decision Logic
- No plan exists → NEXT_AGENT: planning
- Plan exists but no research → NEXT_AGENT: research
- Plan + research exist but no prompt → NEXT_AGENT: prompt_generator
- Prompt generated → NEXT_AGENT: end
## Output Format
NEXT_AGENT: <agent_name>
REASON: <explanation>
MESSAGE: <message_to_agent>
## Tools Available
- execute_host: Run commands on host system
- execute_host: Run kubectl commands on host (use sparingly, research agent handles this)
- execute_bash: Run commands in container
Use tools only for simple verification. Delegate complex work to specialized agents.
Limit iterations to 2 maximum. Keep workflow simple: planning → research → prompt_generator → end.
"""
@@ -59,19 +61,16 @@ def orchestrator_node(state: AgentState) -> AgentState:
iteration_count = state.get("iteration_count", 0)
# 컨텍스트 구성
context_parts = [f"현재 반복 횟수: {iteration_count}/3"]
context_parts = [f"현재 반복 횟수: {iteration_count}/2"]
if state.get("task_plan"):
context_parts.append(f"작업 계획: {state['task_plan']}")
context_parts.append(f"✅ 계획 수립 완료")
if state.get("research_data"):
context_parts.append(f"수집된 정보: {state['research_data']}")
context_parts.append(f"✅ 클러스터 분석 완료")
if state.get("code_outputs"):
context_parts.append(f"생성된 코드: {state['code_outputs']}")
if state.get("review_feedback"):
context_parts.append(f"리뷰 피드백: {state['review_feedback']}")
if state.get("implementation_prompt"):
context_parts.append(f"✅ 구현 프롬프트 생성 완료")
context = "\n".join(context_parts)

50
services/backend/agents/planning_agent.py
View File

@@ -17,32 +17,52 @@ claude_planning = ChatAnthropic(
)
PLANNING_PROMPT = """You are the Planning Agent in a Multi-Agent System.
PLANNING_PROMPT = """You are the K8s Infrastructure Planning Agent.
## Role
Analyze user requests and create actionable task plans.
Analyze user requests for Kubernetes infrastructure and create detailed implementation plans.
## Process
1. Understand what the user wants
2. Classify: backend / frontend / infrastructure / mixed
3. Break down into steps
4. Identify information needed
5. Define success criteria
## Your Mission
When a user wants to deploy something (e.g., "Tekton", "Harbor", "Prometheus"):
1. Understand what they want to deploy
2. Design the folder structure for K8s manifests
3. Plan YAML file organization
4. Identify what K8s resources are needed
5. Determine what information to gather from their cluster
## Output Format (JSON)
```json
{
"task_type": "backend | frontend | infrastructure | mixed",
"summary": "Brief task summary",
"steps": [
{"step": 1, "description": "...", "agent": "research|code_*"}
"task_type": "k8s_infrastructure",
"summary": "Deploy X to Kubernetes cluster",
"target_tool": "Name of the tool/service to deploy",
"folder_structure": {
"base_path": "deploy/X",
"directories": ["base", "overlays/prod", "overlays/dev"],
"files": {
"base/deployment.yaml": "Main deployment manifest",
"base/service.yaml": "Service definition",
"base/kustomization.yaml": "Kustomize base"
}
},
"k8s_resources": [
{"type": "Namespace", "name": "X"},
{"type": "Deployment", "name": "X"},
{"type": "Service", "name": "X-svc"}
],
"research_needed": ["What info to gather"],
"success_criteria": ["How to verify completion"]
"research_needed": [
"Check existing namespaces",
"Verify storage classes available",
"Check current resource quotas"
],
"implementation_steps": [
{"step": 1, "description": "Create namespace and RBAC", "files": ["namespace.yaml", "serviceaccount.yaml"]},
{"step": 2, "description": "Deploy core components", "files": ["deployment.yaml", "service.yaml"]}
]
}
```
Keep plans simple and actionable. Research agent can explore and find things automatically.
Focus on K8s best practices: namespaces, RBAC, resource limits, health checks, and GitOps compatibility.
"""

144
services/backend/agents/prompt_generator_agent.py Normal file
View File

@@ -0,0 +1,144 @@
"""
Prompt Generator Agent (Claude 4.5)
다른 AI에게 전달할 구현 프롬프트를 Markdown으로 생성
"""
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage
from .state import AgentState
import os
import json
# Claude 4.5 모델 초기화
claude_prompt_gen = ChatAnthropic(
model="claude-sonnet-4-20250514",
api_key=os.getenv("ANTHROPIC_API_KEY"),
temperature=0.5
)
PROMPT_GENERATOR_SYSTEM = """You are the Prompt Generator Agent.
## Role
Generate implementation prompts for other AI assistants (like Claude Code, ChatGPT, etc.).
## Input
- Planning data: folder structure, YAML files, K8s resources
- Research data: current cluster state, existing resources
## Output Format (Markdown)
Create a comprehensive prompt that another AI can use to implement the infrastructure:
```markdown
# Deploy [TOOL_NAME] to Kubernetes
## Context
[Brief description of current cluster state from research data]
## Folder Structure
Create the following directory structure:
```
deploy/[tool]/
├── base/
│ ├── deployment.yaml
│ ├── service.yaml
│ └── kustomization.yaml
└── overlays/
└── prod/
└── kustomization.yaml
```
## Implementation Steps
### Step 1: [Title]
**File:** `deploy/[tool]/base/namespace.yaml`
```yaml
[Example YAML with placeholders]
```
### Step 2: [Title]
**File:** `deploy/[tool]/base/deployment.yaml`
```yaml
[Example YAML with specific recommendations]
```
## Key Considerations
- Resource limits: [specific recommendations based on cluster]
- Storage: [based on available StorageClasses]
- Networking: [based on existing services]
- RBAC: [specific permissions needed]
## Validation Commands
```bash
kubectl apply -k deploy/[tool]/overlays/prod
kubectl get pods -n [namespace]
kubectl logs -n [namespace] deployment/[name]
```
## Expected Outcome
[What should be running after implementation]
```
## Guidelines
1. Be specific and actionable
2. Include actual YAML examples (not just descriptions)
3. Reference the cluster's current state from research data
4. Provide validation steps
5. Make it copy-paste ready for another AI
"""
def prompt_generator_node(state: AgentState) -> AgentState:
"""
Prompt Generator 노드: 다른 AI에게 전달할 구현 프롬프트 생성
"""
messages = state["messages"]
task_plan = state.get("task_plan", {})
research_data = state.get("research_data", {})
# 입력 데이터 준비
plan_summary = json.dumps(task_plan, indent=2, ensure_ascii=False) if task_plan else "No plan available"
research_summary = json.dumps(research_data, indent=2, ensure_ascii=False) if research_data else "No research data"
# 사용자 원래 요청
user_request = messages[0]["content"] if messages else "Deploy infrastructure"
print(f"\n{'='*80}")
print(f"Prompt Generator Agent - Generating implementation prompt")
print(f"{'='*80}")
# Claude 호출
response = claude_prompt_gen.invoke([
SystemMessage(content=PROMPT_GENERATOR_SYSTEM),
HumanMessage(content=f"""Generate an implementation prompt for this request:
**User Request:** {user_request}
**Planning Data:**
```json
{plan_summary}
```
**Research Data (Cluster State):**
```json
{research_summary}
```
Create a comprehensive Markdown prompt that another AI can use to implement this infrastructure.
Include specific YAML examples, folder structure, and validation steps.
""")
])
content = response.content
print(f"✅ Prompt generated ({len(content)} characters)")
# 상태 업데이트
state["implementation_prompt"] = content
state["messages"].append({
"role": "prompt_generator",
"content": content
})
state["current_agent"] = "end" # 완료
return state

174
services/backend/agents/review_agent.py
View File

@@ -1,174 +0,0 @@
"""
Review & Test Agent (Claude)
코드 리뷰, 품질 검증, 테스트 전략 수립
"""
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage
from .state import AgentState
from tools.bash_tool import bash_tools
import os
import json
# Claude 모델 초기화 (리뷰는 고품질 모델 사용)
claude_review = ChatAnthropic(
model="claude-sonnet-4-20250514",
api_key=os.getenv("ANTHROPIC_API_KEY"),
temperature=0.3
)
REVIEW_PROMPT = """You are the Review & Test Agent.
## Role
Review code quality, security, and performance. Run tests if needed.
## Tools
- execute_bash: Run tests, linters, builds
- execute_host: Check deployments with kubectl
## Output Format (JSON)
```json
{
"approved": true/false,
"overall_score": 85,
"summary": "Brief assessment",
"issues": [
{"severity": "high|medium|low", "category": "security|performance|quality", "description": "...", "recommendation": "..."}
],
"strengths": ["..."],
"next_steps": ["..."]
}
```
## Approval
- approved: true if no critical issues
- approved: false if major security/quality problems
"""
def review_node(state: AgentState) -> AgentState:
"""
Review 노드: 코드 리뷰 및 품질 검증
"""
messages = state["messages"]
code_outputs = state.get("code_outputs", {})
task_plan = state.get("task_plan", {})
# 코드 리뷰 요청 구성
code_summary = "\n\n".join([
f"### {agent_type.upper()}\n{code}"
for agent_type, code in code_outputs.items()
])
review_request = f"""
작업 계획: {task_plan.get('summary', '')}
성공 기준: {task_plan.get('success_criteria', [])}
생성된 코드:
{code_summary}
위 코드를 검토하고, 필요시 테스트를 실행한 후 JSON 형식으로 피드백을 제공해주세요.
"""
# Claude에 bash 도구 바인딩
claude_with_tools = claude_review.bind_tools(bash_tools)
# Claude 호출
response = claude_with_tools.invoke([
SystemMessage(content=REVIEW_PROMPT),
HumanMessage(content=review_request)
])
# Tool calls 처리
tool_outputs = []
if hasattr(response, 'tool_calls') and response.tool_calls:
for tool_call in response.tool_calls:
try:
tool_name = tool_call.get('name') or tool_call.get('function', {}).get('name', 'unknown')
tool_args_raw = tool_call.get('args') or tool_call.get('function', {}).get('arguments', {})
# tool_args가 문자열인 경우 JSON 파싱 시도
if isinstance(tool_args_raw, str):
try:
tool_args = json.loads(tool_args_raw)
except json.JSONDecodeError:
# JSON 파싱 실패 시 빈 딕셔너리 사용
tool_args = {}
print(f"⚠️ Failed to parse tool_args as JSON: {tool_args_raw}")
elif isinstance(tool_args_raw, dict):
tool_args = tool_args_raw
else:
tool_args = {}
print(f"⚠️ Unexpected tool_args type: {type(tool_args_raw)}")
# tool_name에 따라 올바른 도구 선택
from tools.bash_tool import execute_bash, execute_host
if tool_name == "execute_host":
tool_func = execute_host
else:
tool_func = execute_bash
# 필수 파라미터 확인
if 'command' not in tool_args:
tool_outputs.append(f"\n❌ **{tool_name}** failed: 'command' parameter is required")
continue
tool_result = tool_func.invoke(tool_args)
tool_outputs.append(f"\n🔧 **Review {tool_name}({tool_args.get('command', '')[:50]}...)**:\n{tool_result}")
except Exception as e:
error_detail = str(e)
import traceback
print(f"❌ Tool call error: {error_detail}")
print(traceback.format_exc())
tool_outputs.append(f"\n❌ **{tool_name if 'tool_name' in locals() else 'unknown'}** failed: {error_detail}")
# Tool 결과와 함께 재호출
if tool_outputs:
tool_context = "\n".join(tool_outputs)
response = claude_review.invoke([
SystemMessage(content=REVIEW_PROMPT),
HumanMessage(content=review_request),
HumanMessage(content=f"도구 실행 결과:\n{tool_context}\n\n이제 JSON 형식으로 최종 리뷰를 제공해주세요.")
])
content = response.content
if tool_outputs:
content = "\n".join(tool_outputs) + "\n\n" + content
# JSON 파싱 시도
try:
if "```json" in content:
json_str = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
json_str = content.split("```")[1].split("```")[0].strip()
else:
json_str = content
review_feedback = json.loads(json_str)
except Exception as e:
review_feedback = {
"approved": False,
"overall_score": 50,
"summary": "리뷰 파싱 실패",
"issues": [{"severity": "high", "category": "quality", "description": f"JSON 파싱 에러: {str(e)}", "recommendation": "재검토 필요"}],
"strengths": [],
"next_steps": ["피드백 재생성"]
}
# 상태 업데이트
state["review_feedback"] = review_feedback
state["is_approved"] = review_feedback.get("approved", False)
state["messages"].append({
"role": "review",
"content": content
})
# 승인되지 않았고 반복 횟수가 3 미만이면 재작업
if not state["is_approved"] and state["iteration_count"] < 3:
state["iteration_count"] += 1
state["current_agent"] = "orchestrator" # 재작업을 위해 orchestrator로
else:
state["current_agent"] = "end" # 승인되었거나 최대 반복 도달
return state

17
services/backend/agents/state.py
View File

@@ -1,17 +1,16 @@
"""
공유 상태 정의 (AgentState)
K8s 인프라 분석 및 계획 수립에 특화
"""
from typing import TypedDict, Optional
class AgentState(TypedDict):
"""에이전트 간 공유되는 상태"""
messages: list # 대화 메시지 이력
current_agent: str # 현재 활성 에이전트
task_plan: Optional[dict] # Planning Agent 출력
research_data: Optional[dict] # Research Agent 출력
code_outputs: dict # Code Agent(s) 출력
review_feedback: Optional[dict] # Review Agent 출력
iteration_count: int # 반복 횟수 (최대 3회)
is_approved: bool # 최종 승인 여부
error: Optional[str] # 에러 메시지
messages: list # 대화 메시지 이력
current_agent: str # 현재 활성 에이전트
task_plan: Optional[dict] # Planning Agent 출력 (폴더 구조, YAML 설계)
research_data: Optional[dict] # Research Agent 출력 (K8s 클러스터 상태)
implementation_prompt: Optional[str] # Prompt Generator 출력 (Markdown 프롬프트)
iteration_count: int # 반복 횟수 (최대 2회)
error: Optional[str] # 에러 메시지

63
services/backend/chainlit_app.py
View File

@@ -22,20 +22,20 @@ except:
async def start():
"""채팅 시작 시"""
await cl.Message(
content="🤖 **Multi-Agent System v2.0**에 오신 것을 환영합니다!\n\n"
"저는 다음 전문가 팀과 함께 작업합니다:\n\n"
"**계획 & 조율**\n"
content="☸️ **K8s Infrastructure Planning System v3.0**에 오신 것을 환영합니다!\n\n"
"당신의 Kubernetes 클러스터 상태를 분석하고 인프라 배포 계획을 수립해드립니다.\n\n"
"**에이전트 팀**\n"
"- 🎼 **Orchestrator** (Claude 4.5): 전체 워크플로우 조율\n"
"- 📋 **Planning Agent** (Claude 4.5): 작업 계획 수립\n\n"
"**정보 수집**\n"
"- 🔍 **Research Agent** (Groq): 정보 수집 및 분석\n\n"
"**코드 작성**\n"
"- ⚙️ **Backend Agent** (Groq): 백엔드 개발\n"
"- 🎨 **Frontend Agent** (Groq): 프론트엔드 개발\n"
"- 🏗️ **Infrastructure Agent** (Groq): K8s/DevOps\n\n"
"**품질 보증**\n"
"- ✅ **Review Agent** (Claude): 코드 리뷰 & 테스트\n\n"
"무엇을 도와드릴까요?"
"- 📋 **Planning Agent** (Claude 4.5): 폴더 구조 & YAML 설계\n"
"- 🔍 **Research Agent** (Groq): K8s 클러스터 상태 분석\n"
"- 📝 **Prompt Generator** (Claude 4.5): 구현 가이드 생성\n\n"
"**사용 예시**\n"
"```\n"
"Tekton을 도입하고 싶어\n"
"Harbor를 배포하려고 해\n"
"Prometheus를 설치하고 싶어\n"
"```\n\n"
"배포하고 싶은 도구를 알려주세요!"
).send()
@@ -57,10 +57,8 @@ async def main(message: cl.Message):
"current_agent": "orchestrator",
"task_plan": None,
"research_data": None,
"code_outputs": {},
"review_feedback": None,
"implementation_prompt": None,
"iteration_count": 0,
"is_approved": False,
"error": None
}
@@ -81,27 +79,20 @@ async def main(message: cl.Message):
agent_content = last_message["content"]
# 사용자에게 보여줄 에이전트만 필터링
user_facing_agents = ["planning", "research", "backend_developer",
"frontend_developer", "infrastructure_engineer", "review"]
user_facing_agents = ["planning", "research", "prompt_generator"]
if agent_name in user_facing_agents:
# 에이전트별 아이콘
agent_icons = {
"planning": "📋",
"research": "🔍",
"backend_developer": "⚙️",
"frontend_developer": "🎨",
"infrastructure_engineer": "🏗️",
"review": "✅"
"prompt_generator": "📝"
}
agent_display_names = {
"planning": "계획 수립",
"research": "정보 수집",
"backend_developer": "백엔드 개발",
"frontend_developer": "프론트엔드 개발",
"infrastructure_engineer": "인프라 구성",
"review": "코드 리뷰"
"planning": "인프라 계획 수립",
"research": "클러스터 상태 분석",
"prompt_generator": "구현 가이드 생성"
}
icon = agent_icons.get(agent_name, "🤖")
@@ -124,13 +115,10 @@ async def main(message: cl.Message):
# Orchestrator는 간단한 상태 메시지만 표시
current_agent = state.get("current_agent", "")
status_icons = {
"planning": "📋 계획 수립 중...",
"research": "🔍 정보 수집 중...",
"code_backend": "⚙️ 백엔드 코드 작성 중...",
"code_frontend": "🎨 프론트엔드 코드 작성 중...",
"code_infrastructure": "🏗️ 인프라 구성 중...",
"review": "✅ 코드 검토 중...",
"end": "✨ 완료!"
"planning": "📋 인프라 계획 수립 중...",
"research": "🔍 클러스터 상태 분석 중...",
"prompt_generator": "📝 구현 가이드 생성 중...",
"end": "✨ 완료! 아래 프롬프트를 복사하여 사용하세요."
}
status_text = status_icons.get(current_agent, "⏳ 작업 중...")
status_msg.content = status_text
@@ -164,10 +152,7 @@ def rename(orig_author: str):
"orchestrator": "Orchestrator (Claude 4.5)",
"planning": "Planning Agent (Claude 4.5)",
"research": "Research Agent (Groq)",
"backend_developer": "Backend Agent (Groq)",
"frontend_developer": "Frontend Agent (Groq)",
"infrastructure_engineer": "Infrastructure Agent (Groq)",
"review": "Review Agent (Claude)"
"prompt_generator": "Prompt Generator (Claude 4.5)"
}
return rename_dict.get(orig_author, orig_author)

55
services/backend/workflow.py
View File

@@ -1,6 +1,6 @@
"""
LangGraph Iterative Workflow
반복적 워크플로우: Planning → Research → Code → Review (최대 3회 반복)
LangGraph K8s Infrastructure Planning Workflow
워크플로우: Planning → Research → Prompt Generation → End
"""
from typing import Literal
from langgraph.graph import StateGraph, END
@@ -9,36 +9,24 @@ from agents import (
orchestrator_node,
planning_node,
research_node,
backend_code_node,
frontend_code_node,
infrastructure_code_node,
review_node
prompt_generator_node
)
def router(state: AgentState) -> Literal[
"planning",
"research",
"code_backend",
"code_frontend",
"code_infrastructure",
"review",
"prompt_generator",
"end"
]:
"""
다음 에이전트 라우팅 로직
K8s 인프라 계획: planning → research → prompt_generator → end
"""
current = state.get("current_agent", "orchestrator")
# 명시적으로 지정된 다음 에이전트로 이동
if current in [
"planning",
"research",
"code_backend",
"code_frontend",
"code_infrastructure",
"review"
]:
if current in ["planning", "research", "prompt_generator"]:
return current
# end 상태
@@ -51,22 +39,18 @@ def router(state: AgentState) -> Literal[
def create_mas_workflow():
"""
MAS Iterative Workflow 생성
K8s Infrastructure Planning Workflow 생성
워크플로우:
User Request
User Request (e.g., "Deploy Tekton")
↓
Orchestrator → Planning → Orchestrator
↓
Research → Orchestrator
Research (K8s cluster analysis) → Orchestrator
↓
Code (Backend/Frontend/Infrastructure) → Orchestrator
Prompt Generator (Markdown implementation guide) → Orchestrator
↓
Review → Orchestrator
↓ (if not approved and iteration < 3)
Research (반복)
↓ (if approved or iteration >= 3)
End
End (User copies prompt to another AI)
"""
workflow = StateGraph(AgentState)
@@ -74,10 +58,7 @@ def create_mas_workflow():
workflow.add_node("orchestrator", orchestrator_node)
workflow.add_node("planning", planning_node)
workflow.add_node("research", research_node)
workflow.add_node("code_backend", backend_code_node)
workflow.add_node("code_frontend", frontend_code_node)
workflow.add_node("code_infrastructure", infrastructure_code_node)
workflow.add_node("review", review_node)
workflow.add_node("prompt_generator", prompt_generator_node)
# 시작점: Orchestrator
workflow.set_entry_point("orchestrator")
@@ -89,10 +70,7 @@ def create_mas_workflow():
{
"planning": "planning",
"research": "research",
"code_backend": "code_backend",
"code_frontend": "code_frontend",
"code_infrastructure": "code_infrastructure",
"review": "review",
"prompt_generator": "prompt_generator",
"end": END
}
)
@@ -100,12 +78,7 @@ def create_mas_workflow():
# 각 에이전트는 작업 후 Orchestrator로 복귀
workflow.add_edge("planning", "orchestrator")
workflow.add_edge("research", "orchestrator")
workflow.add_edge("code_backend", "orchestrator")
workflow.add_edge("code_frontend", "orchestrator")
workflow.add_edge("code_infrastructure", "orchestrator")
# Review는 승인 여부에 따라 처리 (review_node 내부에서 current_agent 설정)
workflow.add_edge("review", "orchestrator")
workflow.add_edge("prompt_generator", "orchestrator")
return workflow.compile()