mirror of
https://github.com/Jerryplusy/AI-powered-switches.git
synced 2025-10-14 09:49:19 +00:00
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from typing import Dict, Any, Coroutine
|
||
|
||
import httpx
|
||
from openai import OpenAI
|
||
import json
|
||
from src.backend.app.utils.exceptions import SiliconFlowAPIException
|
||
from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam
|
||
from src.backend.app.utils.logger import logger
|
||
|
||
|
||
class AIService:
|
||
def __init__(self, api_key: str, api_url: str):
|
||
self.api_key = api_key
|
||
self.api_url = api_url
|
||
self.client = OpenAI(
|
||
api_key=self.api_key,
|
||
base_url=self.api_url,
|
||
# timeout=httpx.Timeout(30.0)
|
||
)
|
||
|
||
async def parse_command(self, command: str) -> Any | None:
|
||
"""
|
||
调用硅基流动API解析中文命令
|
||
"""
|
||
prompt = """
|
||
你是一个网络设备配置专家，精通各种类型的路由器的配置,请将以下用户的中文命令转换为网络设备配置JSON
|
||
但是请注意，由于贪婪的人们追求极高的效率，所以你必须严格按照 JSON 格式返回数据，不要包含任何额外文本或 Markdown 代码块
|
||
返回格式要求：
|
||
1. 必须包含'type'字段指明配置类型(vlan/interface/acl/route等)
|
||
2. 必须包含'commands'字段，包含可直接执行的命令列表
|
||
3. 其他参数根据配置类型动态添加
|
||
4. 不要包含解释性文本、步骤说明或注释
|
||
示例命令：'创建VLAN 100，名称为TEST'
|
||
示例返回：{"type": "vlan", "vlan_id": 100, "name": "TEST", "commands": ["vlan 100", "name TEST"]}
|
||
"""
|
||
|
||
messages = [
|
||
ChatCompletionSystemMessageParam(role="system", content=prompt),
|
||
ChatCompletionUserMessageParam(role="user", content=command)
|
||
]
|
||
|
||
try:
|
||
response = self.client.chat.completions.create(
|
||
model="deepseek-ai/DeepSeek-V3",
|
||
messages=messages,
|
||
temperature=0.3,
|
||
max_tokens=1000,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
|
||
logger.debug(response)
|
||
|
||
config_str = response.choices[0].message.content.strip()
|
||
|
||
try:
|
||
config = json.loads(config_str)
|
||
return config
|
||
except json.JSONDecodeError:
|
||
# 尝试修复可能的多余字符
|
||
if config_str.startswith("```json"):
|
||
config_str = config_str[7:-3].strip()
|
||
return json.loads(config_str)
|
||
raise SiliconFlowAPIException("Invalid JSON format returned from AI")
|
||
except KeyError:
|
||
logger.error(KeyError)
|
||
raise SiliconFlowAPIException("errrrrrrro")
|
||
|
||
except Exception as e:
|
||
raise SiliconFlowAPIException(
|
||
detail=f"API请求失败: {str(e)}",
|
||
status_code=getattr(e, "status_code", 500)
|
||
) |