diff --git a/ruoyi-fastapi-backend/config/env.py b/ruoyi-fastapi-backend/config/env.py index 05ce750..850c8ec 100644 --- a/ruoyi-fastapi-backend/config/env.py +++ b/ruoyi-fastapi-backend/config/env.py @@ -221,6 +221,13 @@ class SearchSettings: SEARCH_CACHE_TTL = int(os.getenv("SEARCH_CACHE_TTL", "1800")) +class DeepSeekSettings: + """DeepSeek大语言模型配置""" + DEEPSEEK_API_BASE = os.getenv("DEEPSEEK_API_BASE", "https://api.deepseek.com") + DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "sk-56b608b26a6949e4b09b5bf5f11c8f5b") + DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat") + + class GetConfig: """ 获取配置 @@ -302,6 +309,11 @@ class GetConfig: def get_search_config(self): """获取搜索配置""" return SearchSettings() + + @lru_cache() + def get_deepseek_config(self): + """获取DeepSeek配置""" + return DeepSeekSettings() @staticmethod def parse_cli_args(): @@ -351,4 +363,6 @@ RAGFlowConfig = get_config.get_ragflow_config() # compreface配置 ComprefaceConfig = get_config.get_compreface_config() # 搜索配置 -SearchConfig = get_config.get_search_config() \ No newline at end of file +SearchConfig = get_config.get_search_config() +# DeepSeek配置 +DeepSeekConfig = get_config.get_deepseek_config() \ No newline at end of file diff --git a/ruoyi-fastapi-backend/module_admin/service/search_service.py b/ruoyi-fastapi-backend/module_admin/service/search_service.py index ff8fb7a..989a972 100644 --- a/ruoyi-fastapi-backend/module_admin/service/search_service.py +++ b/ruoyi-fastapi-backend/module_admin/service/search_service.py @@ -1,7 +1,8 @@ import hashlib import json +import re from datetime import datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from config.env import SearchConfig from utils.log_util import logger @@ -15,13 +16,35 @@ class SearchServiceError(Exception): class SearchService: """搜索服务 - 处理联网搜索请求""" - # 搜索关键词 - 用于判断是否应该使用搜索服务 + # 基础搜索关键词 - 用于判断是否应该使用搜索服务 SEARCH_KEYWORDS = ( - '搜索', '查找', '新闻', '时事', '最新', '今天', '现在', '当前', - '天气', '温度', '气温', '下雨', '晴', '多云', - 'search', 'find', 'news', 'latest', 'today', 'current', 'now', - 'weather', 'temperature', 'rain', 'sunny', 'cloudy' + # 操作类 + '搜索', '查找', '查询', '搜一下', '帮我查', '帮我搜', '检索', + # 时间类 + '最新', '最近', '今日', '今天', '现在', '当前', '昨日', '昨天', + # 天气类 + '天气', '温度', '气温', '降水', '下雨', '下雪', '晴', '多云', '阴天', '雾霾', + # 新闻资讯类 + '新闻', '时事', '资讯', '动态', '报道', '消息', + # 事件类 + '发生了什么', '怎么回事', '事件', '事故', '活动', + # 英文关键词 + 'search', 'find', 'query', 'latest', 'recent', 'today', 'now', 'current', + 'weather', 'temperature', 'news', 'events', 'what happened' ) + + # 领域特定关键词（扩展） + DOMAIN_KEYWORDS = { + '科技': ('AI', '人工智能', '区块链', '元宇宙', '量子计算', '5G', '芯片'), + '体育': ('足球', '篮球', '奥运会', '世界杯', 'NBA', '英超', '比分'), + '娱乐': ('电影', '电视剧', '明星', '演唱会', '综艺', '票房'), + '财经': ('股票', '基金', '汇率', '利率', '财经', '股市', '行情'), + '健康': ('疫情', '疫苗', '健康', '医疗', '疾病', '症状'), + '教育': ('考试', '考研', '高考', '录取', '学校', '教育'), + } + + # 否定关键词（这些词出现时不应使用搜索） + NEGATIVE_KEYWORDS = ('不要搜索', '不搜索', '无需搜索', '不用搜索', 'don\'t search', 'no search') _client = SearchAPIClient( base_url=SearchConfig.SEARCH_API_BASE, @@ -40,8 +63,42 @@ class SearchService: """ if not question: return False + + # 转换为小写用于匹配 normalized = question.strip().lower() - return any(k in question for k in cls.SEARCH_KEYWORDS) or any(k in normalized for k in cls.SEARCH_KEYWORDS) + + # 检查是否包含否定关键词 + if any(neg_k in normalized for neg_k in cls.NEGATIVE_KEYWORDS): + return False + + # 1. 基础关键词匹配 + if any(k in question for k in cls.SEARCH_KEYWORDS) or any(k in normalized for k in cls.SEARCH_KEYWORDS): + return True + + # 2. 领域特定关键词匹配（需要同时满足领域词+时间/操作词） + for domain, domain_terms in cls.DOMAIN_KEYWORDS.items(): + # 检查是否包含领域关键词 + has_domain_term = any(term in question for term in domain_terms) or any(term.lower() in normalized for term in domain_terms) + + # 检查是否包含时间或操作关键词 + has_time_or_action = any( + k in question for k in ('最新', '最近', '今日', '今天', '搜索', '查找', '查询') + ) or any( + k in normalized for k in ('latest', 'recent', 'today', 'search', 'find', 'query') + ) + + if has_domain_term and has_time_or_action: + return True + + # 3. 模式匹配（如包含时间+事件的模式） + # 示例："2023年发生了什么"、"昨天的新闻"等 + time_patterns = r'(\d{4}年|\d{1,2}月\d{1,2}日|昨天|昨日|今天|今日|最近\d+天|最近\d+周|最近\d+月)' + if re.search(time_patterns, question) and any( + k in question for k in ('发生', '新闻', '事件', '动态') + ): + return True + + return False @classmethod async def get_search_answer(cls, question: str, redis=None) -> Dict[str, Any]: @@ -133,6 +190,9 @@ class SearchService: # 提取答案框（如果有） answer_box = payload.get('answer_box', {}) + # 提取知识图谱（如果有） + knowledge_graph = payload.get('knowledge_graph', {}) + # 构建结果列表 results = [] for item in organic_results[:SearchConfig.SEARCH_NUM_RESULTS]: @@ -140,27 +200,39 @@ class SearchService: 'title': item.get('title', ''), 'link': item.get('link', ''), 'snippet': item.get('snippet', ''), + 'source': item.get('source', ''), } results.append(result_item) # 构建用户友好的消息 message_parts = [f'关于"{query}"的搜索结果：\n'] - # 如果有答案框，优先显示 + # 如果有知识图谱，优先显示 + if knowledge_graph: + kg_title = knowledge_graph.get('title', '') + kg_description = knowledge_graph.get('description', '') + if kg_title and kg_description: + simplified_desc = cls._simplify_text(kg_description) + message_parts.append(f'【知识卡片】{kg_title}: {simplified_desc}\n') + + # 如果有答案框，显示答案 if answer_box: answer = answer_box.get('answer') or answer_box.get('snippet', '') if answer: - message_parts.append(f'【快速答案】{answer}\n') + simplified_answer = cls._simplify_text(answer) + message_parts.append(f'【快速答案】{simplified_answer}\n') # 添加搜索结果 if results: - message_parts.append('\n【相关结果】') - for idx, item in enumerate(results, 1): + message_parts.append('\n【相关信息】') + for idx, item in enumerate(results[:3], 1): # 只显示前3条 title = item['title'] snippet = item['snippet'] - message_parts.append(f'\n{idx}. {title}') - if snippet: - message_parts.append(f' {snippet}') + if title: + message_parts.append(f'\n{idx}. {title}') + if snippet: + simplified_snippet = cls._simplify_text(snippet) + message_parts.append(f' {simplified_snippet}') else: message_parts.append('\n未找到相关结果。') @@ -168,39 +240,107 @@ class SearchService: 'type': 'search', 'query': query, 'answer_box': answer_box, + 'knowledge_graph': knowledge_graph, 'results': results, 'message': '\n'.join(message_parts), - 'context_for_llm': cls._prepare_llm_context(query, answer_box, results), + 'context_for_llm': cls._prepare_llm_context(query, answer_box, knowledge_graph, results), 'raw': payload, } + @staticmethod + def _simplify_text(text: str) -> str: + """简化文本内容，去除冗余信息，保留核心内容 + + Args: + text: 原始文本 + + Returns: + 简化后的文本 + """ + if not text: + return "" + + # 去除多余空格和换行 + text = re.sub(r'\s+', ' ', text.strip()) + + # 去除冗余的标点符号 + text = re.sub(r'[，,、；;]+', '，', text) + text = re.sub(r'[。.!]+', '。', text) + + # 去除括号内的辅助信息（可选） + # text = re.sub(r'\([^)]*\)|\[[^\]]*\]|\{[^}]*\}', '', text) + + # 限制长度（根据需要调整） + max_length = 150 + if len(text) > max_length: + # 尝试在句子边界截断 + sentence_endings = re.finditer(r'[。.!?]', text[:max_length]) + if sentence_endings: + last_end = list(sentence_endings)[-1].end() + text = text[:last_end] + "..." + else: + text = text[:max_length] + "..." + + return text + @classmethod - def _prepare_llm_context(cls, query: str, answer_box: Dict[str, Any], results: list) -> str: + def _prepare_llm_context(cls, query: str, answer_box: Dict[str, Any], knowledge_graph: Dict[str, Any], results: List[Dict[str, Any]]) -> str: """准备用于LLM处理的简洁上下文 Args: query: 用户查询 answer_box: 答案框数据 + knowledge_graph: 知识图谱数据 results: 搜索结果列表 Returns: 简洁的上下文文本，供LLM生成答案 """ context_parts = [] + context_tokens = 0 + max_tokens = 500 # 限制上下文长度，避免超过模型token限制 + + # 添加知识图谱信息（如果有） + if knowledge_graph: + kg_title = knowledge_graph.get('title', '') + kg_description = knowledge_graph.get('description', '') + if kg_title and kg_description: + simplified_desc = cls._simplify_text(kg_description) + kg_context = f"知识卡片：{kg_title} - {simplified_desc}" + context_parts.append(kg_context) + context_tokens += len(kg_context) # 添加答案框信息（如果有） - if answer_box: + if answer_box and context_tokens < max_tokens: answer = answer_box.get('answer') or answer_box.get('snippet', '') if answer: - context_parts.append(f"快速答案：{answer}") + simplified_answer = cls._simplify_text(answer) + answer_context = f"快速答案：{simplified_answer}" + context_parts.append(answer_context) + context_tokens += len(answer_context) - # 添加前3条最相关的搜索结果 - if results: + # 添加搜索结果 + if results and context_tokens < max_tokens: context_parts.append("\n相关信息：") - for idx, item in enumerate(results[:3], 1): + for idx, item in enumerate(results, 1): + if context_tokens >= max_tokens: + break + snippet = item.get('snippet', '') if snippet: - # 只保留摘要，去掉链接和标题以节省token - context_parts.append(f"{idx}. {snippet}") + simplified_snippet = cls._simplify_text(snippet) + result_context = f"{idx}. {simplified_snippet}" + context_parts.append(result_context) + context_tokens += len(result_context) + + # 最多添加3条结果 + if idx >= 3: + break - return "\n".join(context_parts) if context_parts else "未找到相关信息" + final_context = "\n".join(context_parts) if context_parts else "未找到相关信息" + + # 最终检查长度 + if len(final_context) > max_tokens * 2: # 粗略估算token数 + final_context = final_context[:max_tokens * 2] + "..." + + return final_context diff --git a/ruoyi-fastapi-backend/requirements.txt b/ruoyi-fastapi-backend/requirements.txt index 2ac675b..e3e6e7c 100644 --- a/ruoyi-fastapi-backend/requirements.txt +++ b/ruoyi-fastapi-backend/requirements.txt @@ -17,3 +17,4 @@ requests==2.32.3 SQLAlchemy[asyncio]==2.0.38 sqlglot[rs]==26.6.0 user-agents==2.2.0 +openai==1.37.1 diff --git a/ruoyi-fastapi-backend/test_full_flow.py b/ruoyi-fastapi-backend/test_full_flow.py deleted file mode 100644 index 0ce3276..0000000 --- a/ruoyi-fastapi-backend/test_full_flow.py +++ /dev/null @@ -1,149 +0,0 @@ -""" -测试搜索+LLM完整流程的脚本 - -使用方法： -1. 确保服务器正在运行 -2. 运行此脚本：python test_full_flow.py -""" -import asyncio -import json -import sys - -try: - import httpx -except ImportError: - print("错误: 缺少 httpx 模块") - print("请运行: pip install httpx 或使用系统已安装的httpx") - sys.exit(1) - - -async def test_search_llm_flow(): - """测试完整的搜索+LLM流程""" - - print("=" * 60) - print("搜索+LLM完整流程测试") - print("=" * 60) - - # 服务器配置 - base_url = "http://localhost:9099/dev-api" - - # 测试问题 - test_question = "今天有什么大事发生" - - print(f"\n测试问题: {test_question}") - print("-" * 60) - - # 首先需要登录获取token - print("\n步骤1: 登录获取token...") - - # 这里需要您提供登录凭证 - print("\n⚠️ 需要登录凭证才能测试") - print("请提供以下信息：") - username = input("用户名: ").strip() or "admin" - password = input("密码: ").strip() or "admin123" - - login_url = f"{base_url}/login" - login_data = { - "username": username, - "password": password - } - - try: - async with httpx.AsyncClient(timeout=30.0) as client: - # 登录 - print(f"\n正在登录...") - response = await client.post(login_url, json=login_data) - - if response.status_code != 200: - print(f"✗ 登录失败: HTTP {response.status_code}") - print(response.text) - return - - result = response.json() - if result.get('code') != 200: - print(f"✗ 登录失败: {result.get('msg', '未知错误')}") - return - - token = result.get('data', {}).get('access_token') - if not token: - print("✗ 未获取到token") - return - - print(f"✓ 登录成功，token: {token[:20]}...") - - # 测试对话接口 - print(f"\n步骤2: 调用对话接口...") - - # 需要chat_id，这里使用一个测试ID - chat_id = input("\n请输入chat_id (直接回车使用默认): ").strip() - if not chat_id: - print("⚠️ 需要有效的chat_id才能调用RAGFlow") - print("请在RAGFlow中创建一个聊天助手，并提供其chat_id") - return - - converse_url = f"{base_url}/system/ragflow/converse_with_chat_assistant" - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json" - } - converse_data = { - "chat_id": chat_id, - "question": test_question, - "stream": False - } - - print(f"\n发送请求到: {converse_url}") - print(f"问题: {test_question}") - print(f"Chat ID: {chat_id}") - - response = await client.post(converse_url, json=converse_data, headers=headers) - - print(f"\n响应状态码: {response.status_code}") - - if response.status_code == 200: - result = response.json() - print("\n✓ 请求成功!") - print("\n" + "=" * 60) - print("响应内容:") - print("=" * 60) - print(json.dumps(result, ensure_ascii=False, indent=2)) - - # 提取答案 - if result.get('code') == 200: - data = result.get('data', {}) - answer = data.get('answer') or data.get('content') or data.get('message', '') - if answer: - print("\n" + "=" * 60) - print("【机器人回答】") - print("=" * 60) - print(answer) - print("=" * 60) - else: - print(f"\n⚠️ 业务错误: {result.get('msg', '未知错误')}") - else: - print(f"\n✗ 请求失败") - print(response.text) - - except httpx.RequestError as e: - print(f"\n✗ 网络错误: {e}") - except Exception as e: - print(f"\n✗ 未知错误: {type(e).__name__}: {e}") - import traceback - traceback.print_exc() - - print("\n" + "=" * 60) - print("测试完成") - print("=" * 60) - - -if __name__ == "__main__": - print("\n提示：此测试需要以下条件：") - print("1. ✓ 搜索服务API密钥已配置") - print("2. ? FastAPI服务器正在运行 (python server.py)") - print("3. ? RAGFlow服务正在运行") - print("4. ? 有效的用户账号和chat_id") - print("\n如果服务器未运行，请先在另一个终端运行: python server.py") - - input("\n按回车继续测试...") - - asyncio.run(test_search_llm_flow()) diff --git a/ruoyi-fastapi-backend/test_search_api.py b/ruoyi-fastapi-backend/test_search_api.py deleted file mode 100644 index 9e831a5..0000000 --- a/ruoyi-fastapi-backend/test_search_api.py +++ /dev/null @@ -1,108 +0,0 @@ -""" -简单的搜索API测试脚本 - 直接测试HTTP请求 -""" -import asyncio -import os -from dotenv import load_dotenv - -# 加载环境变量 -load_dotenv('.env.dev') - -try: - import httpx -except ImportError: - print("错误: 缺少 httpx 模块") - print("请运行: pip install httpx") - exit(1) - - -async def test_search_api(): - """直接测试SerpAPI""" - - api_key = os.getenv('SEARCH_API_KEY', '') - - print("=" * 60) - print("搜索API测试") - print("=" * 60) - print(f"API Key: {'已配置' if api_key else '未配置'}") - - if not api_key: - print("\n错误: SEARCH_API_KEY 未配置") - print("请在 .env.dev 中设置 SEARCH_API_KEY") - return - - # 测试查询 - query = "北京天气" - - print(f"\n测试查询: {query}") - print("-" * 60) - - # 构建请求 - url = "https://serpapi.com/search" - params = { - 'q': query, - 'api_key': api_key, - 'engine': 'google', - 'hl': 'zh-cn', - 'gl': 'cn', - 'num': 5, - } - - try: - async with httpx.AsyncClient(timeout=15.0) as client: - print("发送请求...") - response = await client.get(url, params=params) - - print(f"状态码: {response.status_code}") - - if response.status_code == 200: - data = response.json() - - print("\n✓ 搜索成功!") - - # 显示答案框 - answer_box = data.get('answer_box', {}) - if answer_box: - answer = answer_box.get('answer') or answer_box.get('snippet', '') - if answer: - print(f"\n【快速答案】") - print(answer) - - # 显示搜索结果 - organic_results = data.get('organic_results', []) - if organic_results: - print(f"\n【搜索结果】({len(organic_results)}条)") - for idx, item in enumerate(organic_results[:3], 1): - print(f"\n{idx}. {item.get('title', 'N/A')}") - print(f" {item.get('link', 'N/A')}") - snippet = item.get('snippet', '') - if snippet: - print(f" {snippet[:100]}...") - - # 显示API使用情况 - search_metadata = data.get('search_metadata', {}) - if search_metadata: - print(f"\n【API信息】") - print(f"总耗时: {search_metadata.get('total_time_taken', 'N/A')}秒") - print(f"请求ID: {search_metadata.get('id', 'N/A')}") - - else: - print(f"\n✗ 请求失败") - try: - error_data = response.json() - print(f"错误: {error_data.get('error', response.text)}") - except: - print(f"错误: {response.text}") - - except httpx.RequestError as e: - print(f"\n✗ 网络错误: {e}") - except Exception as e: - print(f"\n✗ 未知错误: {type(e).__name__}: {e}") - - print("\n" + "=" * 60) - print("测试完成") - print("=" * 60) - - -if __name__ == "__main__": - asyncio.run(test_search_api()) diff --git a/ruoyi-fastapi-backend/test_search_llm_standalone.py b/ruoyi-fastapi-backend/test_search_llm_standalone.py index 5a99e5a..6796214 100644 --- a/ruoyi-fastapi-backend/test_search_llm_standalone.py +++ b/ruoyi-fastapi-backend/test_search_llm_standalone.py @@ -24,7 +24,6 @@ async def test_search_with_llm(): # 导入服务 from module_admin.service.search_service import SearchService, SearchServiceError - from utils.ragflow_client_manager import init_ragflow_client, get_ragflow_client # 测试问题 test_question = "今天有什么大事发生" @@ -61,90 +60,49 @@ async def test_search_with_llm(): print(f"\n✗ 搜索失败: {e}") return - # 步骤3: 初始化RAGFlow客户端 - print("\n步骤3: 初始化RAGFlow客户端...") + # 步骤3: 初始化DeepSeek客户端 + print("\n步骤3: 初始化DeepSeek客户端...") try: - await init_ragflow_client() - client = await get_ragflow_client() - print("✓ RAGFlow客户端初始化成功") + from utils.deepseek_client import DeepSeekAPIClient + client = DeepSeekAPIClient() + print("✓ DeepSeek客户端初始化成功") except Exception as e: - print(f"\n✗ RAGFlow初始化失败: {e}") - print("\n⚠️ 需要配置RAGFlow服务才能测试LLM处理") + print(f"\n✗ DeepSeek初始化失败: {e}") + print("\n⚠️ 需要配置DeepSeek服务才能测试LLM处理") print("请在 .env.dev 中配置:") - print(" RAGFLOW_BASE_URL=http://your_ragflow_server") - print(" RAGFLOW_API_KEY=your_api_key") + print(" DEEPSEEK_API_KEY=your_deepseek_api_key") return - # 步骤4: 构建增强提示词 - print("\n步骤4: 构建LLM提示词...") - enhanced_question = f"""请基于以下搜索结果，简洁准确地回答用户的问题。 - -用户问题：{test_question} - -搜索结果： -{context} - -要求： -1. 直接回答问题，不要说"根据搜索结果"等前缀 -2. 答案要简洁准确，突出重点 -3. 如果是天气查询，只返回当前天气状况和温度 -4. 如果是新闻查询，只返回最重要的新闻摘要 -5. 用自然的语气回答，像人类助手一样""" + # 步骤4: 准备传递给DeepSeek的内容 + print("\n步骤4: 准备DeepSeek输入...") - print("✓ 提示词构建完成") - print(f"\n【提示词预览】") - print(enhanced_question[:300] + '...') - - # 步骤5: 调用RAGFlow LLM - print("\n步骤5: 调用RAGFlow LLM生成答案...") - print("⚠️ 需要提供chat_id才能调用LLM") - - chat_id = input("\n请输入chat_id (直接回车跳过LLM测试): ").strip() - - if not chat_id: - print("\n✓ 跳过LLM测试") - print("\n" + "=" * 70) - print("测试总结") - print("=" * 70) - print("✓ 搜索服务工作正常") - print("✓ 上下文提取成功") - print("⚠️ 未测试LLM处理（需要chat_id）") - print("\n要完整测试，请:") - print("1. 在RAGFlow中创建聊天助手") - print("2. 获取chat_id") - print("3. 重新运行此脚本并输入chat_id") - return + # 步骤5: 调用DeepSeek LLM + print("\n步骤5: 调用DeepSeek LLM生成答案...") try: - # 调用RAGFlow - result = await client.converse_with_chat_assistant( - chat_id=chat_id, - question=enhanced_question, - stream=False + # 调用DeepSeek API + answer = await client.chat( + question=test_question, + context=context, + temperature=0.7, + max_tokens=1024 ) print("✓ LLM调用成功!") - # 提取答案 - if result.get('code') == 0: - data = result.get('data', {}) - answer = data.get('answer', '') - - print("\n" + "=" * 70) - print("【机器人最终回答】") - print("=" * 70) - print(answer) - print("=" * 70) - - print("\n✓ 完整流程测试成功!") - print("\n流程总结:") - print(" 1. 检测到搜索关键词 → 触发搜索服务") - print(" 2. 调用SerpAPI → 获取搜索结果") - print(" 3. 提取关键信息 → 构建LLM上下文") - print(" 4. 调用RAGFlow LLM → 生成自然语言答案") - print(" 5. 返回简洁答案 → 用户体验优化") - else: - print(f"\n⚠️ LLM返回错误: {result.get('msg', '未知错误')}") + print("\n" + "=" * 70) + print("【机器人最终回答】") + print("=" * 70) + print(answer) + print("=" * 70) + + print("\n✓ 完整流程测试成功!") + print("\n流程总结:") + print(" 1. 检测到搜索关键词 → 触发搜索服务") + print(" 2. 调用SerpAPI → 获取搜索结果") + print(" 3. 提取关键信息 → 构建LLM上下文") + print(" 4. 调用DeepSeek LLM → 生成自然语言答案") + print(" 5. 返回简洁答案 → 用户体验优化") except Exception as e: print(f"\n✗ LLM调用失败: {e}") @@ -159,8 +117,7 @@ async def test_search_with_llm(): if __name__ == "__main__": print("\n提示：此测试需要以下配置：") print("1. ✓ SEARCH_API_KEY (搜索服务)") - print("2. ✓ RAGFLOW_BASE_URL 和 RAGFLOW_API_KEY (LLM服务)") - print("3. ? chat_id (可选，用于LLM测试)") + print("2. ✓ DEEPSEEK_API_KEY (LLM服务)") print() try: diff --git a/ruoyi-fastapi-backend/test_search_service.py b/ruoyi-fastapi-backend/test_search_service.py deleted file mode 100644 index 7111664..0000000 --- a/ruoyi-fastapi-backend/test_search_service.py +++ /dev/null @@ -1,83 +0,0 @@ -""" -搜索服务测试脚本 -""" -import asyncio -import sys -import os - -# 添加项目路径到Python路径 -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from module_admin.service.search_service import SearchService, SearchServiceError - - -async def test_search_service(): - """测试搜索服务""" - - print("=" * 60) - print("搜索服务测试") - print("=" * 60) - - # 测试用例 - test_queries = [ - "北京今天天气怎么样", - "最新的人工智能新闻", - "搜索Python编程", - ] - - for query in test_queries: - print(f"\n{'='*60}") - print(f"测试查询: {query}") - print(f"{'='*60}") - - # 检查是否应该处理 - should_handle = SearchService.should_handle(query) - print(f"✓ 关键词检测: {'通过' if should_handle else '未通过'}") - - if should_handle: - try: - # 获取搜索结果（不使用Redis缓存） - result = await SearchService.get_search_answer(query, redis=None) - - print(f"\n✓ 搜索成功!") - print(f"\n类型: {result.get('type')}") - print(f"查询: {result.get('query')}") - - # 显示答案框（如果有） - answer_box = result.get('answer_box', {}) - if answer_box: - answer = answer_box.get('answer') or answer_box.get('snippet', '') - if answer: - print(f"\n【快速答案】\n{answer}") - - # 显示搜索结果 - results = result.get('results', []) - if results: - print(f"\n【搜索结果】({len(results)}条)") - for idx, item in enumerate(results[:3], 1): # 只显示前3条 - print(f"\n{idx}. {item.get('title', 'N/A')}") - print(f" 链接: {item.get('link', 'N/A')}") - snippet = item.get('snippet', '') - if snippet: - print(f" 摘要: {snippet[:100]}...") - - # 显示格式化消息 - print(f"\n【格式化消息】") - message = result.get('message', '') - # 只显示前500个字符 - print(message[:500] + ('...' if len(message) > 500 else '')) - - except SearchServiceError as e: - print(f"\n✗ 搜索失败: {e}") - except Exception as e: - print(f"\n✗ 未知错误: {type(e).__name__}: {e}") - - print() - - print("=" * 60) - print("测试完成") - print("=" * 60) - - -if __name__ == "__main__": - asyncio.run(test_search_service()) diff --git a/ruoyi-fastapi-backend/utils/deepseek_client.py b/ruoyi-fastapi-backend/utils/deepseek_client.py new file mode 100644 index 0000000..3068960 --- /dev/null +++ b/ruoyi-fastapi-backend/utils/deepseek_client.py @@ -0,0 +1,137 @@ +from openai import AsyncOpenAI +from typing import Any, Dict, List, Optional +from config.env import DeepSeekConfig +from utils.log_util import logger + + +class DeepSeekAPIError(Exception): + """DeepSeek API调用异常""" + + def __init__(self, status_code: int, message: str): + self.status_code = status_code + self.message = message + super().__init__(f"DeepSeek API error {status_code}: {message}") + + +class DeepSeekAPIClient: + """DeepSeek API客户端 - 用于调用DeepSeek大语言模型 + 使用官方推荐的OpenAI SDK进行调用 + """ + + def __init__(self, base_url: str = None, api_key: str = None, timeout: float = 30.0): + self.base_url = base_url or DeepSeekConfig.DEEPSEEK_API_BASE + self.base_url = self.base_url.rstrip('/') if self.base_url else 'https://api.deepseek.com' + self.api_key = api_key or DeepSeekConfig.DEEPSEEK_API_KEY + self.timeout = timeout + self.model = DeepSeekConfig.DEEPSEEK_MODEL + + # 初始化OpenAI客户端 + if not self.api_key: + raise DeepSeekAPIError(401, 'DeepSeek API密钥未配置') + + self.client = AsyncOpenAI( + api_key=self.api_key, + base_url=self.base_url, + timeout=self.timeout + ) + + async def chat_completion( + self, + messages: List[Dict[str, str]], + model: str = None, + temperature: float = 0.7, + max_tokens: int = 1024, + stream: bool = False, + ) -> Dict[str, Any]: + """调用DeepSeek聊天补全API + + Args: + messages: 聊天消息列表，格式为 [{"role": "user", "content": "你的问题"}] + model: 模型名称，默认使用配置中的模型 + temperature: 温度参数，控制生成内容的随机性 + max_tokens: 最大生成token数 + stream: 是否流式返回 + + Returns: + API返回的JSON响应 + + Raises: + DeepSeekAPIError: API调用失败时抛出 + """ + model = model or self.model + + try: + response = await self.client.chat.completions.create( + model=model, + messages=messages, + temperature=temperature, + max_tokens=max_tokens, + stream=stream + ) + + # 转换为字典格式返回，保持与原有接口兼容 + if not stream: + return response.model_dump() + else: + return response # 流式响应直接返回 + + except Exception as exc: + logger.error(f"DeepSeek API调用失败: {str(exc)}") + raise DeepSeekAPIError(500, str(exc)) from exc + + async def chat( + self, + question: str, + context: str = None, + model: str = None, + temperature: float = 0.7, + max_tokens: int = 1024, + ) -> str: + """简化的聊天接口 + + Args: + question: 用户问题 + context: 上下文信息（可选） + model: 模型名称 + temperature: 温度参数 + max_tokens: 最大生成token数 + + Returns: + 生成的回答 + + Raises: + DeepSeekAPIError: API调用失败时抛出 + """ + # 构建消息列表 + messages = [] + + # 如果有上下文，添加到系统消息中 + if context: + messages.append({ + "role": "system", + "content": f"基于以下上下文信息回答用户问题：\n{context}" + }) + + # 添加用户问题 + messages.append({ + "role": "user", + "content": question + }) + + # 调用API + response = await self.chat_completion( + messages=messages, + model=model, + temperature=temperature, + max_tokens=max_tokens, + stream=False + ) + + # 提取回答 + answer = response.get('choices', [{}])[0].get('message', {}).get('content', '') + + if not answer: + logger.error(f"DeepSeek API返回空回答: {response}") + raise DeepSeekAPIError(200, 'DeepSeek API返回空回答') + + return answer.strip()