mirror of
https://github.com/Jerryplusy/AI-powered-switches.git
synced 2025-10-14 09:49:19 +00:00
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
from typing import Any, List, Dict
|
||
from openai import AsyncOpenAI
|
||
import json
|
||
from src.backend.app.utils.exceptions import SiliconFlowAPIException
|
||
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam
|
||
|
||
class AIService:
|
||
def __init__(self, api_key: str, api_url: str):
|
||
self.client = AsyncOpenAI(api_key=api_key, base_url=api_url)
|
||
|
||
async def parse_command(self, command: str, devices: List[Dict]) -> Dict[str, Any]:
|
||
"""
|
||
针对一组设备和一条自然语言命令，生成每台设备的配置 JSON
|
||
"""
|
||
devices_str = json.dumps(devices, ensure_ascii=False, indent=2)
|
||
|
||
example = """[{"device": {"name": "sw1","ip": "192.168.1.10","vendor": "huawei","username": "NONE", "password": "Huawei"},"config": {"type": "vlan","vlan_id": 300,"name": "Sales","commands": ["system-view","vlan 300","name Sales","quit","quit","save","Y"]}}]"""
|
||
|
||
prompt = f"""
|
||
你是一个网络设备配置专家。现在有以下设备：
|
||
{devices_str}
|
||
|
||
用户输入了一条命令：{command}
|
||
|
||
你的任务：
|
||
- 为每台设备分别生成配置
|
||
- 输出一个 JSON 数组，每个元素对应一台设备
|
||
- 每个对象必须包含:
|
||
- device: 原始设备信息 (name, ip, vendor,username,password)
|
||
- config: 配置详情
|
||
- type: 配置类型 (如 vlan/interface/acl/route)
|
||
- commands: 可直接执行的命令数组 (必须包含进入配置、退出、保存命令)
|
||
- 其他字段: 根据配置类型动态添加
|
||
- 严格返回 JSON，不要包含解释说明或 markdown
|
||
|
||
各厂商保存命令规则：
|
||
- 华为: system-view → quit → save Y
|
||
- 思科: enable → configure terminal → exit → write memory
|
||
- H3C: system-view → quit → save
|
||
- 锐捷: enable → configure terminal → exit → write
|
||
- 中兴: enable → configure terminal → exit → write memory
|
||
|
||
返回示例（仅作为格式参考，不要照抄 VLAN ID 和命令内容，请根据实际命令生成）：{example}
|
||
"""
|
||
|
||
messages = [
|
||
ChatCompletionSystemMessageParam(role="system", content=prompt),
|
||
ChatCompletionUserMessageParam(role="user", content=command)
|
||
]
|
||
|
||
try:
|
||
response = await self.client.chat.completions.create(
|
||
model="deepseek-ai/DeepSeek-V3",
|
||
messages=messages,
|
||
temperature=0.2,
|
||
max_tokens=1500,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
|
||
config_str = response.choices[0].message.content.strip()
|
||
configs = json.loads(config_str)
|
||
|
||
return {"success": True, "results": configs}
|
||
|
||
except Exception as e:
|
||
raise SiliconFlowAPIException(
|
||
detail=f"AI 解析配置失败: {str(e)}",
|
||
status_code=getattr(e, "status_code", 500)
|
||
)
|