This commit is contained in:
zhanggangyong
2026-01-15 17:22:16 +00:00
parent 3a19aacd43
commit f065b00d12
8 changed files with 741 additions and 32 deletions
+7
View File
@@ -2,10 +2,17 @@ FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# 复制应用代码
COPY requirements.txt .
COPY app.py .
COPY k8s_manager.py .
COPY database.py .
COPY agent_manager/ ./agent_manager/
# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt
BIN
View File
Binary file not shown.
Binary file not shown.
+250 -19
View File
@@ -198,8 +198,10 @@ class CreateAgentRequest(BaseModel):
"""创建Agent请求(旧版)"""
name: str = Field(..., description="Agent名称", min_length=1, max_length=63)
template: str = Field(..., description="模板类型")
framework: Optional[str] = Field(default="API", description="Agent框架类型: MCP, A2A, API")
config: Dict = Field(default_factory=dict, description="配置信息")
env: Optional[Dict[str, str]] = Field(default_factory=dict, description="环境变量")
namespace: Optional[str] = Field(default=None, description="Kubernetes命名空间,默认使用环境变量NAMESPACE的值")
class AgentResponse(BaseModel):
@@ -207,6 +209,7 @@ class AgentResponse(BaseModel):
name: str
namespace: str
status: str
framework: Optional[str] = None
created_at: Optional[str] = None
template: Optional[str] = None
service_port: Optional[int] = None
@@ -240,6 +243,7 @@ class PodStatusResponse(BaseModel):
status: str
health_status: Optional[str] = None # 新增:健康状态 (healthy, unhealthy, degraded)
template: Optional[str] = None
framework: Optional[str] = None # Agent框架类型 (MCP, A2A, API)
created_at: Optional[str] = None
node: Optional[str] = None
pod_ip: Optional[str] = None
@@ -249,6 +253,8 @@ class PodStatusResponse(BaseModel):
access_url: Optional[str] = None
endpoints: Optional[Dict] = None
conditions: Optional[List[Dict]] = None
# 访问信息(从数据库读取)
access_info: Optional[Dict] = None # 包含 external_ip, domain, URLs 等
class PodMetricsResponse(BaseModel):
@@ -279,15 +285,16 @@ async def root():
@app.post("/agents", response_model=AgentResponse)
async def create_agent(request: CreateAgentRequest):
async def create_agent(request: CreateAgentRequest, db: Session = Depends(get_db)):
"""
创建AI Agent Pod
创建AI Agent Pod(在独立命名空间中,并创建Service和Ingress)
Args:
request: 创建请求(name, template, config, user_id可选)
request: 创建请求(name, template, config, namespace可选, user_id可选)
db: 数据库会话
Returns:
创建的Agent信息包括pod_id
创建的Agent信息包括pod_id、service和ingress信息
"""
try:
logger.info(f"收到创建Agent请求: {request.name}, 模板: {request.template}")
@@ -300,8 +307,18 @@ async def create_agent(request: CreateAgentRequest):
detail=f"无效的模板类型。支持的模板: {', '.join(valid_templates)}"
)
# 验证 framework 类型
valid_frameworks = ["MCP", "A2A", "API"]
framework = (request.framework or "API").upper()
if framework not in valid_frameworks:
raise HTTPException(
status_code=400,
detail=f"无效的框架类型。支持的框架: {', '.join(valid_frameworks)}"
)
# 合并环境变量到config
config_data = request.config.copy()
config_data["agent_framework"] = framework # 添加框架类型到配置
if request.env:
config_data["env"] = request.env
logger.info(f"环境变量: {list(request.env.keys())}")
@@ -312,36 +329,180 @@ async def create_agent(request: CreateAgentRequest):
config_data["labels"] = {}
config_data["labels"]["user-id"] = user_id
config_data["labels"]["managed-by"] = "agent-manager"
config_data["labels"]["app"] = request.name
config_data["labels"]["framework"] = framework.lower() # 添加框架标签
# 创建Pod
result = k8s_manager.create_pod(
# 步骤1: 为每个Agent创建独立的命名空间
agent_namespace = k8s_manager.create_agent_namespace(
agent_name=request.name,
owner_id=user_id
)
logger.info(f"✅ Agent {request.name} 将部署在独立命名空间: {agent_namespace}")
# 步骤2: 创建Pod(在独立命名空间中)
temp_manager = K8sManager(namespace=agent_namespace, kubeconfig_path=KUBECONFIG_PATH)
result = temp_manager.create_pod(
pod_name=request.name,
template=request.template,
config_data=config_data
)
# 步骤3: 获取服务端口
service_port = k8s_manager.TEMPLATE_PORTS.get(request.template)
# 步骤4: 创建 LoadBalancer Service(AKS 会自动分配外网 IP)
service_info = None
dns_info = None
if service_port:
try:
import time
time.sleep(2) # 等待Pod启动
# 创建 LoadBalancer Service
service_info = temp_manager.create_service(
service_name=f"{request.name}-service",
namespace=agent_namespace,
pod_selector={"app": request.name},
service_port=80, # 外部访问端口
target_port=service_port # Pod内部端口
)
logger.info(f"✅ LoadBalancer Service 创建成功: {service_info['name']}")
# 步骤5: 等待 LoadBalancer IP 分配并创建 DNS
try:
logger.info("等待 LoadBalancer 外网 IP 分配...")
external_ip = temp_manager.wait_for_loadbalancer_ip(
service_name=f"{request.name}-service",
namespace=agent_namespace,
max_wait=120, # 最多等待2分钟
interval=5
)
service_info["external_ip"] = external_ip
logger.info(f"✅ LoadBalancer 外网 IP: {external_ip}")
# 步骤6: 自动创建 Azure DNS 记录
try:
logger.info(f"创建 DNS 记录: {request.name}.taijiagnet.com")
dns_info = temp_manager.create_dns_record(
subdomain=request.name,
ip_address=external_ip
)
logger.info(f"✅ DNS 记录: {dns_info['domain']} -> {external_ip}")
except Exception as e:
logger.warning(f"DNS 记录创建失败: {str(e)}")
dns_info = {"status": "failed", "error": str(e)}
except Exception as e:
logger.warning(f"等待 LoadBalancer IP 超时: {str(e)}")
logger.info(" 外网 IP 将在后台继续分配")
except Exception as e:
logger.warning(f"创建 LoadBalancer Service 失败: {str(e)}")
# 获取 Pod 详细信息(包括 pod_id)
try:
import time
time.sleep(1) # 等待 Pod 创建完成
pod = k8s_manager.v1.read_namespaced_pod(
pod = temp_manager.v1.read_namespaced_pod(
name=request.name,
namespace=NAMESPACE
namespace=agent_namespace
)
result["pod_id"] = pod.metadata.uid
result["pod_ip"] = pod.status.pod_ip
result["host_ip"] = pod.status.host_ip
result["node_name"] = pod.spec.node_name
result["namespace"] = agent_namespace # 更新为实际使用的命名空间
result["framework"] = framework # 添加框架类型到响应
result["owner_info"] = {
"user_id": user_id,
"agent_name": request.name,
"namespace": NAMESPACE,
"namespace": agent_namespace,
"framework": framework,
"labels": pod.metadata.labels
}
logger.info(f"✅ Agent创建成功,Pod ID: {result['pod_id']}, 用户: {user_id}")
# 添加 LoadBalancer Service 信息到响应
if service_info:
result["service_info"] = service_info
# 构建访问信息
external_ip = service_info.get("external_ip")
if external_ip:
# 有外网 IP
result["access_info"] = {
"external_ip": external_ip,
"ip_url": f"http://{external_ip}:80",
"service_url": f"http://{service_info['cluster_ip']}:80",
"pod_url": f"http://{pod.status.pod_ip}:{service_port}" if pod.status.pod_ip and service_port else None
}
# 添加 DNS 信息
if dns_info and dns_info.get("status") == "created":
result["dns_info"] = dns_info
result["access_info"]["domain"] = dns_info["domain"]
result["access_info"]["domain_url"] = f"http://{dns_info['domain']}"
result["access_info"]["recommended"] = f"http://{dns_info['domain']}"
logger.info(f" - 推荐访问: http://{dns_info['domain']}")
else:
result["access_info"]["recommended"] = f"http://{external_ip}:80"
logger.info(f" - 外网访问: http://{external_ip}:80")
else:
# IP 还在分配中
result["access_info"] = {
"status": "pending",
"external_ip": None,
"note": "LoadBalancer IP 正在分配中,请稍后查询"
}
logger.info(f" - 外网 IP 正在分配中")
logger.info(f"✅ Agent创建成功!")
logger.info(f" - Pod ID: {result['pod_id']}")
logger.info(f" - Namespace: {agent_namespace}")
logger.info(f" - User: {user_id}")
except Exception as e:
logger.warning(f"获取Pod详细信息失败: {str(e)}")
# 保存到数据库
try:
# 提取访问信息
access_info = result.get("access_info", {})
external_ip = access_info.get("external_ip")
domain = access_info.get("domain")
ip_url = access_info.get("ip_url")
domain_url = access_info.get("domain_url")
recommended_url = access_info.get("recommended", domain_url or ip_url)
# 创建Agent记录
db_agent = Agent(
name=request.name,
display_name=request.name,
owner_id=user_id,
agent_type=AgentType.PLATFORM, # 默认为平台类型
status=AgentStatus.RUNNING,
agent_framework=framework.lower(),
namespace=agent_namespace,
service_name=service_info.get("name") if service_info else None,
external_ip=external_ip,
domain=domain,
ip_url=ip_url,
domain_url=domain_url,
recommended_url=recommended_url,
min_replicas=1,
max_replicas=1,
current_replicas=1
)
db.add(db_agent)
db.commit()
db.refresh(db_agent)
logger.info(f"✅ Agent信息已保存到数据库: {db_agent.id}")
except Exception as e:
logger.error(f"保存Agent到数据库失败: {str(e)}")
# 不抛出异常,因为Agent已经在K8s中创建成功
return AgentResponse(**result)
except Exception as e:
@@ -350,23 +511,48 @@ async def create_agent(request: CreateAgentRequest):
@app.delete("/agents/{agent_name}", response_model=MessageResponse)
async def delete_agent(agent_name: str):
async def delete_agent(agent_name: str, db: Session = Depends(get_db)):
"""
删除AI Agent Pod
删除AI Agent(包括独立命名空间、LoadBalancer Service、DNS 记录和数据库记录)
Args:
agent_name: Agent名称
db: 数据库会话
Returns:
删除结果
"""
try:
logger.info(f"收到删除Agent请求: {agent_name}")
result = k8s_manager.delete_pod(pod_name=agent_name)
# 步骤1: 删除 DNS 记录
try:
dns_result = k8s_manager.delete_dns_record(subdomain=agent_name)
if dns_result.get("status") == "deleted":
logger.info(f"✅ DNS 记录已删除: {dns_result.get('domain')}")
except Exception as e:
logger.warning(f"删除 DNS 记录失败(可忽略): {str(e)}")
# 步骤2: 删除 Agent 的独立命名空间(会自动删除Pod、Service等所有资源)
result = k8s_manager.delete_agent_namespace(agent_name=agent_name)
if result.get("status") == "not_found":
raise HTTPException(status_code=404, detail=result.get("message"))
# 步骤3: 从数据库删除Agent记录
try:
db_agent = db.query(Agent).filter(Agent.name == agent_name).first()
if db_agent:
db.delete(db_agent)
db.commit()
logger.info(f"✅ Agent数据库记录已删除: {agent_name}")
else:
logger.warning(f"⚠️ Agent {agent_name} 在数据库中未找到")
except Exception as db_error:
logger.error(f"删除数据库记录失败: {str(db_error)}")
# 不抛出异常,因为K8s资源已经删除
logger.info(f"✅ Agent {agent_name} 及其所有资源删除成功")
return MessageResponse(**result)
except HTTPException:
@@ -377,23 +563,68 @@ async def delete_agent(agent_name: str):
@app.get("/agents/{agent_name}/status", response_model=PodStatusResponse)
async def get_agent_status(agent_name: str):
async def get_agent_status(agent_name: str, db: Session = Depends(get_db)):
"""
获取Agent状态
获取Agent详细状态(包括访问信息)
Args:
agent_name: Agent名称
db: 数据库会话
Returns:
Agent状态信息
Agent状态信息(包括Pod状态和访问信息)
"""
try:
logger.info(f"获取Agent状态: {agent_name}")
# 从Kubernetes获取Pod状态
result = k8s_manager.get_pod_status(pod_name=agent_name)
if result.get("status") == "not_found":
raise HTTPException(status_code=404, detail=result.get("message"))
# 从数据库获取Agent记录(包含访问信息)
try:
db_agent = db.query(Agent).filter(Agent.name == agent_name).first()
if db_agent:
# 添加框架类型
result["framework"] = db_agent.agent_framework.upper() if db_agent.agent_framework else "API"
# 添加访问信息
access_info = {}
if db_agent.external_ip:
access_info["external_ip"] = db_agent.external_ip
if db_agent.ip_url:
access_info["ip_url"] = db_agent.ip_url
if db_agent.domain:
access_info["domain"] = db_agent.domain
if db_agent.domain_url:
access_info["domain_url"] = db_agent.domain_url
if db_agent.recommended_url:
access_info["recommended_url"] = db_agent.recommended_url
if db_agent.service_name:
access_info["service_name"] = db_agent.service_name
# 只有当有访问信息时才添加
if access_info:
result["access_info"] = access_info
logger.info(f"✅ 已添加访问信息: domain={db_agent.domain}, ip={db_agent.external_ip}")
else:
logger.warning(f"⚠️ Agent {agent_name} 在数据库中没有访问信息")
else:
logger.warning(f"⚠️ Agent {agent_name} 在数据库中未找到,可能是在数据库启用前创建的")
except Exception as db_error:
logger.error(f"从数据库读取访问信息失败: {str(db_error)}")
# 不抛出异常,继续返回Pod状态信息
return PodStatusResponse(**result)
except HTTPException:
@@ -458,7 +689,7 @@ async def list_templates():
Returns:
模板列表及其配置信息
"""
valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent"]
valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"]
templates_info = []
for template in valid_templates:
@@ -480,7 +711,7 @@ async def list_platform_templates():
平台提供的Agent模板列表
"""
# 平台 Agent 是预定义的标准模板
platform_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "jina_search_agent", "azure_blob_agent"]
platform_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"]
templates_info = []
for template in platform_templates:
@@ -530,7 +761,7 @@ async def get_template_info(template_name: str):
Returns:
模板详细信息(端口、所需环境变量等)
"""
valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent"]
valid_templates = ["echo_agent", "chat_agent", "code_agent", "search_agent", "mysql_agent", "postgresql_agent", "jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a"]
if template_name not in valid_templates:
raise HTTPException(
+12 -8
View File
@@ -14,13 +14,10 @@ from sqlalchemy.orm import sessionmaker, relationship, Session
import enum
import os
# Database URL from environment or default to SQLite
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./agent_manager.db")
# Database URL - PostgreSQL (hardcoded)
DATABASE_URL = "postgresql://taiji:By%40123456.@taijipda.postgres.database.azure.com:5432/taijiagnet"
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {}
)
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
@@ -97,8 +94,8 @@ class Agent(Base):
name = Column(String(100), unique=True, nullable=False, index=True)
display_name = Column(String(200))
# Template reference
template_id = Column(Integer, ForeignKey("templates.id"), nullable=False)
# Template reference (nullable for backward compatibility with non-template agents)
template_id = Column(Integer, ForeignKey("templates.id"), nullable=True)
template = relationship("Template", back_populates="agents")
# Ownership and organization
@@ -149,6 +146,13 @@ class Agent(Base):
service_url = Column(String(500))
namespace = Column(String(100), default="ai-agents")
# Access information (NEW) - LoadBalancer and DNS details
external_ip = Column(String(100), nullable=True) # LoadBalancer external IP
domain = Column(String(500), nullable=True) # Full DNS domain (e.g., agent-name.taijiagnet.com)
ip_url = Column(String(500), nullable=True) # HTTP URL via IP (http://x.x.x.x:80)
domain_url = Column(String(500), nullable=True) # HTTP URL via domain
recommended_url = Column(String(500), nullable=True) # Recommended access URL
# Metadata
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+4 -4
View File
@@ -2,7 +2,7 @@ apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-manager
namespace: default
namespace: agent-manager
labels:
app: agent-manager
spec:
@@ -69,13 +69,13 @@ apiVersion: v1
kind: Service
metadata:
name: agent-manager
namespace: default
namespace: agent-manager
labels:
app: agent-manager
spec:
type: ClusterIP
type: LoadBalancer
ports:
- port: 8000
- port: 80
targetPort: 8000
protocol: TCP
name: http
+432
View File
@@ -6,6 +6,8 @@ from kubernetes.client.rest import ApiException
from typing import Dict, List, Optional
import logging
import os
import requests
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@@ -14,6 +16,14 @@ logger = logging.getLogger(__name__)
class K8sManager:
"""Kubernetes资源管理器"""
# Azure DNS 配置(从环境变量读取)
AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "263c3ff6-1be5-4141-8308-b188464fb297")
AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "c5ba26db-f180-425f-bac3-93708d853988")
AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "ydt8Q~DIOfKXDUOJkzfgaNVNh2d1VrbeIaYuKbpD")
AZURE_SUBSCRIPTION_ID = os.getenv("AZURE_SUBSCRIPTION_ID", "c6c47e4c-f5f4-49f8-b26f-7728862c17d6")
AZURE_RESOURCE_GROUP = os.getenv("AZURE_RESOURCE_GROUP", "taiji-ai-v0")
AZURE_DNS_ZONE = os.getenv("AZURE_DNS_ZONE", "taijiagnet.com")
def __init__(self, namespace: str = "ai-agents", kubeconfig_path: str = None):
"""
初始化Kubernetes管理器
@@ -27,6 +37,7 @@ class K8sManager:
self.v1 = client.CoreV1Api()
self.apps_v1 = client.AppsV1Api()
self.networking_v1 = client.NetworkingV1Api()
# 确保命名空间存在
self._ensure_namespace()
@@ -73,8 +84,240 @@ class K8sManager:
else:
raise
def create_agent_namespace(self, agent_name: str, owner_id: str = None) -> str:
"""为 Agent 创建独立的命名空间
Args:
agent_name: Agent 名称
owner_id: 所有者 ID
Returns:
创建的命名空间名称
"""
# 生成命名空间名称(使用 agent-{agent_name} 格式)
namespace_name = f"agent-{agent_name}"
# 确保命名空间名称符合 DNS 标准(最多 63 个字符,只能包含小写字母、数字和连字符)
namespace_name = namespace_name[:63].lower().strip('-')
try:
# 检查命名空间是否已存在
self.v1.read_namespace(name=namespace_name)
logger.info(f"命名空间 {namespace_name} 已存在")
return namespace_name
except ApiException as e:
if e.status == 404:
# 创建命名空间
labels = {
"managed-by": "agent-manager",
"agent-name": agent_name
}
if owner_id:
labels["owner-id"] = owner_id
namespace_manifest = client.V1Namespace(
metadata=client.V1ObjectMeta(
name=namespace_name,
labels=labels
)
)
self.v1.create_namespace(body=namespace_manifest)
logger.info(f"✅ 创建命名空间 {namespace_name}")
# 复制 ACR secret 到新命名空间
self._copy_acr_secret_to_namespace(namespace_name)
return namespace_name
else:
raise
def _copy_acr_secret_to_namespace(self, target_namespace: str):
"""复制 ACR secret 到目标命名空间
Args:
target_namespace: 目标命名空间
"""
try:
# 从 agent-manager 命名空间读取 acr-secret
source_secret = self.v1.read_namespaced_secret(
name="acr-secret",
namespace="agent-manager"
)
# 创建新的 secret(去除自动生成的字段)
new_secret = client.V1Secret(
metadata=client.V1ObjectMeta(
name="acr-secret",
namespace=target_namespace
),
data=source_secret.data,
type=source_secret.type
)
# 在目标命名空间创建 secret
self.v1.create_namespaced_secret(
namespace=target_namespace,
body=new_secret
)
logger.info(f"✅ 已复制 ACR secret 到命名空间 {target_namespace}")
except ApiException as e:
if e.status == 404:
logger.warning(f"⚠️ 源 ACR secret 不存在,跳过复制")
elif e.status == 409:
logger.info(f"ACR secret 已存在于命名空间 {target_namespace}")
else:
logger.error(f"❌ 复制 ACR secret 失败: {e}")
def create_service(self, service_name: str, namespace: str, pod_selector: Dict[str, str],
service_port: int, target_port: int) -> Dict:
"""为 Agent Pod 创建 LoadBalancer Service
Args:
service_name: Service 名称
namespace: 命名空间
pod_selector: Pod 选择器标签
service_port: Service 端口
target_port: Pod 目标端口
Returns:
创建的 Service 信息(包含外网 IP)
"""
try:
service_manifest = client.V1Service(
metadata=client.V1ObjectMeta(
name=service_name,
namespace=namespace,
labels={"managed-by": "agent-manager"}
),
spec=client.V1ServiceSpec(
selector=pod_selector,
ports=[
client.V1ServicePort(
name="http",
protocol="TCP",
port=service_port,
target_port=target_port
)
],
type="LoadBalancer"
)
)
response = self.v1.create_namespaced_service(
namespace=namespace,
body=service_manifest
)
logger.info(f"✅ LoadBalancer Service {service_name} 创建成功 (namespace: {namespace})")
# 获取外网 IP(可能需要等待分配)
external_ip = None
if response.status.load_balancer.ingress:
lb_ingress = response.status.load_balancer.ingress[0]
external_ip = lb_ingress.ip or lb_ingress.hostname
return {
"name": response.metadata.name,
"namespace": response.metadata.namespace,
"cluster_ip": response.spec.cluster_ip,
"port": service_port,
"external_ip": external_ip,
"type": "LoadBalancer"
}
except ApiException as e:
logger.error(f"创建 Service 失败: {e}")
raise Exception(f"创建 Service 失败: {e.reason}")
def create_ingress(self, ingress_name: str, namespace: str, service_name: str,
service_port: int, host: str = None, path: str = "/") -> Dict:
"""为 Agent Service 创建 Ingress
Args:
ingress_name: Ingress 名称
namespace: 命名空间
service_name: 后端 Service 名称
service_port: Service 端口
host: 域名(可选,如果为 None 则使用默认)
path: 路径前缀
Returns:
创建的 Ingress 信息
"""
try:
# 构建路径规则
http_ingress_path = client.V1HTTPIngressPath(
path=path,
path_type="Prefix",
backend=client.V1IngressBackend(
service=client.V1IngressServiceBackend(
name=service_name,
port=client.V1ServiceBackendPort(number=service_port)
)
)
)
# 构建规则
ingress_rule = client.V1IngressRule(
http=client.V1HTTPIngressRuleValue(paths=[http_ingress_path])
)
# 如果指定了 host,添加到规则中
if host:
ingress_rule.host = host
# 创建 Ingress manifest
ingress_manifest = client.V1Ingress(
metadata=client.V1ObjectMeta(
name=ingress_name,
namespace=namespace,
labels={"managed-by": "agent-manager"},
annotations={
"nginx.ingress.kubernetes.io/rewrite-target": "/",
"nginx.ingress.kubernetes.io/ssl-redirect": "false"
}
),
spec=client.V1IngressSpec(
ingress_class_name="nginx", # 使用 nginx ingress controller
rules=[ingress_rule]
)
)
response = self.networking_v1.create_namespaced_ingress(
namespace=namespace,
body=ingress_manifest
)
logger.info(f"✅ Ingress {ingress_name} 创建成功 (namespace: {namespace})")
# 获取 Ingress IP/域名
ingress_url = None
if response.status.load_balancer.ingress:
lb_ingress = response.status.load_balancer.ingress[0]
if lb_ingress.ip:
ingress_url = f"http://{lb_ingress.ip}{path}"
elif lb_ingress.hostname:
ingress_url = f"http://{lb_ingress.hostname}{path}"
return {
"name": response.metadata.name,
"namespace": response.metadata.namespace,
"host": host,
"path": path,
"url": ingress_url,
"note": "Ingress URL 将在负载均衡器配置完成后可用"
}
except ApiException as e:
logger.error(f"创建 Ingress 失败: {e}")
raise Exception(f"创建 Ingress 失败: {e.reason}")
# 模板端口映射
TEMPLATE_PORTS = {
"echo_agent": 8000,
"chat_agent": 8000,
"code_agent": 8000,
"search_agent": 8000,
"mysql_agent": 8000,
"postgresql_agent": 8000,
"jina_search_agent": 8080,
"azure_blob_agent": 8080,
"azure_blob_agent_mcp": 8080,
@@ -397,6 +640,195 @@ class K8sManager:
logger.error(f"删除Pod失败: {e}")
raise Exception(f"删除Pod失败: {e.reason}")
def delete_agent_namespace(self, agent_name: str) -> Dict:
"""删除Agent的命名空间及其所有资源
Args:
agent_name: Agent 名称
Returns:
删除结果
"""
namespace_name = f"agent-{agent_name}".lower().strip('-')[:63]
try:
# 删除命名空间会自动删除其中的所有资源(Pod、Service、Ingress等)
self.v1.delete_namespace(
name=namespace_name,
body=client.V1DeleteOptions()
)
logger.info(f"✅ 命名空间 {namespace_name} 及其所有资源删除成功")
return {
"status": "success",
"message": f"Agent {agent_name} 的命名空间 {namespace_name} 及所有相关资源已删除",
"namespace": namespace_name
}
except ApiException as e:
if e.status == 404:
return {
"status": "not_found",
"message": f"命名空间 {namespace_name} 不存在"
}
logger.error(f"删除命名空间失败: {e}")
raise Exception(f"删除命名空间失败: {e.reason}")
def _get_azure_access_token(self) -> str:
"""获取 Azure Access Token
Returns:
Azure 访问令牌
"""
try:
token_url = f"https://login.microsoftonline.com/{self.AZURE_TENANT_ID}/oauth2/v2.0/token"
data = {
"client_id": self.AZURE_CLIENT_ID,
"client_secret": self.AZURE_CLIENT_SECRET,
"grant_type": "client_credentials",
"scope": "https://management.azure.com/.default"
}
response = requests.post(token_url, data=data, timeout=10)
response.raise_for_status()
token = response.json().get('access_token')
if not token:
raise Exception("未能获取访问令牌")
return token
except Exception as e:
logger.error(f"获取 Azure 访问令牌失败: {e}")
raise
def create_dns_record(self, subdomain: str, ip_address: str, ttl: int = 300) -> Dict:
"""为 IP 地址创建 Azure DNS A 记录
Args:
subdomain: 子域名(不包括主域名)
ip_address: 要指向的 IP 地址
ttl: DNS TTL(秒),默认 300
Returns:
DNS 记录信息
"""
try:
# 获取访问令牌
token = self._get_azure_access_token()
# 构建 DNS API URL
dns_url = (
f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/"
f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/"
f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01"
)
# DNS 记录数据
dns_data = {
"properties": {
"TTL": ttl,
"ARecords": [{"ipv4Address": ip_address}]
}
}
# 创建 DNS 记录
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
response = requests.put(dns_url, json=dns_data, headers=headers, timeout=30)
response.raise_for_status()
full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}"
logger.info(f"✅ DNS 记录创建成功: {full_domain} -> {ip_address}")
return {
"subdomain": subdomain,
"domain": full_domain,
"ip_address": ip_address,
"ttl": ttl,
"status": "created"
}
except requests.exceptions.RequestException as e:
logger.error(f"创建 DNS 记录失败: {e}")
if hasattr(e.response, 'text'):
logger.error(f"错误详情: {e.response.text}")
raise Exception(f"创建 DNS 记录失败: {str(e)}")
def delete_dns_record(self, subdomain: str) -> Dict:
"""删除 Azure DNS A 记录
Args:
subdomain: 要删除的子域名
Returns:
删除结果
"""
try:
# 获取访问令牌
token = self._get_azure_access_token()
# 构建 DNS API URL
dns_url = (
f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/"
f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/"
f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01"
)
headers = {"Authorization": f"Bearer {token}"}
response = requests.delete(dns_url, headers=headers, timeout=30)
if response.status_code == 404:
logger.info(f"DNS 记录 {subdomain} 不存在,跳过删除")
return {"status": "not_found", "subdomain": subdomain}
response.raise_for_status()
full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}"
logger.info(f"✅ DNS 记录删除成功: {full_domain}")
return {
"subdomain": subdomain,
"domain": full_domain,
"status": "deleted"
}
except requests.exceptions.RequestException as e:
logger.error(f"删除 DNS 记录失败: {e}")
raise Exception(f"删除 DNS 记录失败: {str(e)}")
def wait_for_loadbalancer_ip(self, service_name: str, namespace: str,
max_wait: int = 300, interval: int = 5) -> str:
"""等待 LoadBalancer 分配外网 IP
Args:
service_name: Service 名称
namespace: 命名空间
max_wait: 最大等待时间(秒)
interval: 检查间隔(秒)
Returns:
分配的外网 IP 地址
"""
logger.info(f"等待 LoadBalancer IP 分配... (最多等待 {max_wait} 秒)")
elapsed = 0
while elapsed < max_wait:
try:
service = self.v1.read_namespaced_service(service_name, namespace)
if service.status.load_balancer.ingress:
ip = service.status.load_balancer.ingress[0].ip
if ip:
logger.info(f"✅ LoadBalancer IP 已分配: {ip}")
return ip
except Exception as e:
logger.warning(f"查询 Service 状态失败: {e}")
time.sleep(interval)
elapsed += interval
logger.info(f"等待中... ({elapsed}/{max_wait} 秒)")
raise Exception(f"LoadBalancer IP 分配超时({max_wait} 秒)")
def get_pod_status(self, pod_name: str) -> Dict:
"""
获取Pod状态
+36 -1
View File
@@ -119,6 +119,41 @@ try:
)
db.add(jina_template)
# Intelligent Search Agent (Custom with HTTP service)
search_template = Template(
name="search_agent",
display_name="Intelligent Search Agent",
description="智能搜索代理,提供多源搜索和答案生成功能",
agent_type=AgentType.CUSTOM,
agent_framework="langchain",
image="agnettaiji.azurecr.io/ai-agents/search-agent:v1.0",
port=8080,
env_requirements={
"required": {
"LLM_BASE_URL": "LLM API基础URL",
"LLM_API_KEY": "LLM API密钥",
"SERPER_API_KEY": "Serper API密钥(Google搜索)",
"JINA_API_KEY": "Jina API密钥(内容提取和重排序)"
},
"optional": {
"LLM_MODEL": "LLM模型名称,默认xchat52",
"MAX_ITERATIONS": "最大迭代次数,默认3",
"MAX_RESULTS_PER_QUERY": "每次搜索最大结果数,默认10",
"CONTENT_MAX_LENGTH": "内容最大长度,默认5000",
"LOG_LEVEL": "日志级别,默认INFO",
"TIMEOUT": "超时时间(秒),默认30"
}
},
cpu_request="500m",
cpu_limit="1000m",
memory_request="512Mi",
memory_limit="1Gi",
min_replicas=1,
max_replicas=3,
target_cpu_utilization=70
)
db.add(search_template)
# Echo Agent (Platform - simple example)
echo_template = Template(
name="echo_agent",
@@ -139,7 +174,7 @@ try:
db.add(echo_template)
db.commit()
print("✓ Created 4 default templates")
print("✓ Created 5 default templates")
# 创建默认配额
print("\nCreating default quotas...")