""" 大模型调用服务 - 直接调用 OpenAI 兼容接口完成智能体任务 不再使用 openclaw agent 智能体，全部由大模型直接完成 """ import json import time import requests import logging from config import Config logger = logging.getLogger('llm_client') # 默认大模型配置（首次启动自动写入数据库） DEFAULT_LLM_CONFIG = { 'name': '本地Qwen3.6', 'base_url': 'http://192.168.2.7:18003/v1', 'api_key': 'sk-xxxx', 'model_name': 'unsloth/Qwen3.6-27B-Q4_K_M', 'max_context': 262144, 'is_active': 1 } class LLMClient: """大模型客户端 - 管理多个模型配置并调用""" def __init__(self): self._active = None # 缓存当前激活的配置 # ========== 配置管理 ========== def get_all_configs(self): """获取所有模型配置""" from models.database import db return db.get_llm_configs() def get_active_config(self, force=False): """获取当前激活的模型配置""" if self._active and not force: return self._active from models.database import db config = db.get_active_llm_config() if config: self._active = config else: # 无激活配置时使用默认值 self._active = dict(DEFAULT_LLM_CONFIG) return self._active def add_config(self, name, base_url, api_key, model_name, max_context=262144): """新增模型配置""" from models.database import db return db.add_llm_config( name=name, base_url=base_url, api_key=api_key, model_name=model_name, max_context=max_context ) def update_config(self, config_id, **kwargs): """更新模型配置""" from models.database import db db.update_llm_config(config_id, **kwargs) self._active = None # 使缓存失效 def delete_config(self, config_id): """删除模型配置""" from models.database import db return db.delete_llm_config(config_id) def set_active(self, config_id): """切换激活的模型配置""" from models.database import db ok = db.set_active_llm_config(config_id) self._active = None return ok def test_connection(self, config=None): """ 测试模型连接 Args: config: 可选，直接测试指定配置；None 时测试当前激活配置 Returns: (success, message) """ cfg = config or self.get_active_config() try: url = cfg['base_url'].rstrip('/') + '/chat/completions' headers = {'Content-Type': 'application/json'} api_key = cfg.get('api_key', '') if api_key: headers['Authorization'] = f'Bearer {api_key}' payload = { 'model': cfg['model_name'], 'messages': [ {'role': 'user', 'content': 'ping，请只回复pong'} ], 'max_tokens': 16, 'temperature': 0 } resp = requests.post(url, json=payload, headers=headers, timeout=60) if resp.status_code == 200: data = resp.json() reply = data.get('choices', [{}])[0].get('message', {}).get('content', '') return True, f"连接成功: {reply[:50]}" else: return False, f"HTTP {resp.status_code}: {resp.text[:200]}" except Exception as e: return False, str(e) # ========== 调用大模型 ========== def chat(self, messages, temperature=0.3, max_tokens=8192, timeout=600, config=None): """ 调用大模型对话接口 Args: messages: [{'role': 'user'/'system'/'assistant', 'content': '...'}] temperature: 温度 max_tokens: 最大输出token数 timeout: 超时时间（秒） config: 可选，指定使用的模型配置；None 使用当前激活配置 Returns: (success, result) success=True 时 result 为文本内容 success=False 时 result 为错误信息 """ cfg = config or self.get_active_config() try: url = cfg['base_url'].rstrip('/') + '/chat/completions' headers = {'Content-Type': 'application/json'} api_key = cfg.get('api_key', '') if api_key: headers['Authorization'] = f'Bearer {api_key}' payload = { 'model': cfg['model_name'], 'messages': messages, 'temperature': temperature, 'max_tokens': max_tokens } logger.info(f"[LLM] 调用 {cfg['model_name']} @ {cfg['base_url']} | 消息数: {len(messages)} | 输入字符: {sum(len(m.get('content','')) for m in messages)}") resp = requests.post(url, json=payload, headers=headers, timeout=timeout) if resp.status_code != 200: logger.error(f"[LLM] HTTP {resp.status_code}: {resp.text[:300]}") return False, f"大模型接口返回错误 HTTP {resp.status_code}: {resp.text[:300]}" data = resp.json() reply = data.get('choices', [{}])[0].get('message', {}).get('content', '') usage = data.get('usage', {}) logger.info(f"[LLM] 返回 {len(reply)} 字符 | usage: {usage}") return True, reply except requests.exceptions.Timeout: return False, f"大模型调用超时(>{timeout}秒)" except requests.exceptions.ConnectionError as e: return False, f"无法连接大模型服务: {e}" except Exception as e: logger.error(f"[LLM] 调用异常: {e}") return False, str(e) def chat_json(self, messages, temperature=0.1, max_tokens=8192, timeout=600, config=None): """ 调用大模型并解析 JSON 输出 Returns: (success, data_or_error) """ # 追加要求JSON输出的系统提示 sys_prompt = ( "你是一个严格输出JSON的程序化助手。" "你必须只输出一个合法的JSON对象，不要输出任何多余文字、解释或markdown代码块标记。" "确保JSON语法正确，可以被json.loads直接解析。" ) full_messages = [{'role': 'system', 'content': sys_prompt}] + messages ok, result = self.chat(full_messages, temperature=temperature, max_tokens=max_tokens, timeout=timeout, config=config) if not ok: return False, result parsed = self._extract_json(result) if parsed is None: return False, f"大模型输出无法解析为JSON: {result[:300]}" return True, parsed def _extract_json(self, text): """从文本中提取JSON对象""" if not text: return None text = text.strip() # 去掉 markdown 代码块标记 if text.startswith('```'): lines = text.split('\n') # 去掉第一行 ```json 或 ``` lines = lines[1:] # 去掉最后一行 ``` if lines and lines[-1].strip().startswith('```'): lines = lines[:-1] text = '\n'.join(lines).strip() # 直接尝试解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取 {...} 块 import re match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass # 尝试提取 [...] 块 match = re.search(r'\[.*\]', text, re.DOTALL) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass return None # 全局大模型客户端实例 llm_client = LLMClient()