#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 简化版离线整合测试脚本 使用已有数据测试integration模块的不同效果 调用真实AI进行内容整合 """ import os import json import asyncio from typing import Dict, List, Optional, Any from datetime import datetime from pathlib import Path from dataclasses import dataclass # 添加项目根目录到路径 import sys project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) # 导入核心模块 from core.ai.ai_agent import AIAgent from core.config import ConfigManager, AIModelConfig from utils.prompts import PromptTemplate @dataclass class SimpleNote: """简化的笔记数据模型""" note_id: str title: str content: str author: str likes: int comments: int shares: int note_url: str images: List[str] @dataclass class SimpleSearchResult: """简化的搜索结果""" keyword: str notes: List[SimpleNote] total_count: int success: bool class OfflineIntegrationTester: """离线整合测试器""" def __init__(self, data_file: str): """ 初始化测试器 Args: data_file: 离线数据文件路径 """ self.data_file = data_file self.raw_data = None self.processed_notes = [] # 初始化配置管理器和AI代理 self.config_manager = ConfigManager() self.config_manager.load_from_directory("config", server_mode=True) # 获取AI配置并初始化AI代理 ai_config = self.config_manager.get_config('ai_model', AIModelConfig) self.ai_agent = AIAgent(ai_config) # 加载integration prompt模板 self.prompt_template = PromptTemplate( system_prompt_path="resource/prompt/integration/system.txt", user_prompt_path="resource/prompt/integration/user.txt" ) # 加载数据 self._load_data() print(f"✅ 离线测试器初始化完成，加载了 {len(self.processed_notes)} 条笔记数据") print(f"🤖 AI代理已初始化，模型: {ai_config.model}") def _load_data(self): """加载离线数据""" try: with open(self.data_file, 'r', encoding='utf-8') as f: self.raw_data = json.load(f) # 转换为SimpleNote对象 if 'results' in self.raw_data and self.raw_data['results']: result = self.raw_data['results'][0] # 取第一个搜索结果 for note_data in result.get('notes', []): note = SimpleNote( note_id=note_data.get('note_id', ''), title=note_data.get('title', ''), content=note_data.get('content', ''), author=note_data.get('author', ''), likes=note_data.get('like_count', 0), comments=note_data.get('comment_count', 0), shares=note_data.get('share_count', 0), note_url=note_data.get('note_url', ''), images=note_data.get('images', []) ) self.processed_notes.append(note) except Exception as e: print(f"❌ 数据加载失败: {e}") self.processed_notes = [] def get_search_result(self, keyword: str = None) -> SimpleSearchResult: """获取搜索结果""" if not keyword: keyword = self.raw_data['results'][0]['keyword'] if self.raw_data else '测试关键词' return SimpleSearchResult( keyword=keyword, notes=self.processed_notes, total_count=len(self.processed_notes), success=True ) def create_mock_documents(self, content_type: str = "travel_guide") -> Dict[str, str]: """创建模拟文档内容""" mock_docs = { "travel_guide": { "上海馥桂萌宠园游玩攻略.txt": """上海馥桂萌宠园游玩攻略 一、园区简介 馥桂萌宠园位于上海市嘉定区，是一个集萌宠互动、户外休闲、亲子娱乐为一体的综合性主题公园。 二、开放信息 - 开放时间：9:00-17:00（周一至周日） - 门票价格：成人票80元，儿童票50元（1.2米以下免费） - 地址：上海市嘉定区朱桥镇 - 交通：地铁11号线嘉定西站转公交 三、主要区域 1. 萌宠互动区：可与羊驼、袋鼠、小兔子等动物亲密接触 2. 户外游乐区：秋千、滑梯等儿童设施 3. 餐饮休息区：提供简餐和饮品 4. 桂花林区：春秋季节桂花飘香，适合拍照 四、游玩建议 - 建议游玩时间：3-4小时 - 最佳游玩季节：春秋两季 - 携带物品：防晒用品、湿巾、零食 - 注意事项：爱护动物，文明游园""", "萌宠园交通指南.md": """# 馥桂萌宠园交通指南 ## 公共交通 ### 地铁+公交 1. 地铁11号线至嘉定西站 2. 转乘嘉定69路公交至朱桥站 3. 步行约10分钟即可到达 ## 自驾路线 从市区出发： 1. 走外环高速(A20) 2. 转嘉定环线(A5) 3. 在朱桥出口下高速 4. 沿指示牌行驶约5分钟 ## 停车信息 - 园区提供免费停车场 - 停车位约200个 - 节假日建议早到，避免停车位紧张""" }, "product_info": { "门票套餐信息.txt": """门票套餐信息 基础门票： - 成人票：80元/人 - 儿童票：50元/人（3-12岁） - 老年票：60元/人（65岁以上） - 免票：1.2米以下儿童 套餐选择： 1. 家庭套票（2大1小）：180元 2. 亲子套票（1大1小）：120元 3. 团体票（10人以上）：70元/人 增值服务： - 动物喂食包：20元/份 - 拍照服务：50元/次 - 导览服务：100元/团""" }, "user_reviews": { "游客评价汇总.txt": """游客评价汇总 正面评价： ✅ 动物种类丰富，互动性强 ✅ 环境优美，适合拍照 ✅ 工作人员服务态度好 ✅ 停车方便，交通便利 ✅ 价格合理，性价比高 负面反馈： ❌ 节假日人流量大 ❌ 餐饮选择相对有限 ❌ 部分设施需要维护 ❌ 雨天游玩体验一般 综合评分：4.2/5.0""" } } return mock_docs.get(content_type, mock_docs["travel_guide"]) def _format_xhs_notes(self, notes: List[SimpleNote]) -> str: """格式化小红书笔记内容""" if not notes: return "暂无相关笔记内容" formatted_notes = [] for i, note in enumerate(notes, 1): note_content = f"""第{i}条笔记： 标题：{note.title} 作者：{note.author} 内容：{note.content if note.content else '（无文字内容）'} 互动数据：👍{note.likes} 💬{note.comments} 🔄{note.shares} 笔记链接：{note.note_url} """ if note.images: note_content += f"图片数量：{len(note.images)}张\n" formatted_notes.append(note_content) return "\n".join(formatted_notes) def _format_document_content(self, documents: Dict[str, str]) -> str: """格式化文档内容""" if not documents: return "暂无文档内容" formatted_docs = [] for filename, content in documents.items(): doc_content = f"""文档：{filename} 内容： {content} """ formatted_docs.append(doc_content) return "\n".join(formatted_docs) def test_basic_integration(self): """测试基础整合功能""" print("\n" + "="*60) print("🔄 测试基础整合功能") print("="*60) # 获取数据 search_result = self.get_search_result("上海馥桂萌宠园攻略") documents = self.create_mock_documents("travel_guide") print(f"📚 文档处理结果：") print(f" - 文档数量：{len(documents)}") total_doc_length = sum(len(content) for content in documents.values()) print(f" - 总内容长度：{total_doc_length}") print(f"\n📱 小红书数据：") print(f" - 笔记数量：{len(search_result.notes)}") print(f" - 总点赞数：{sum(note.likes for note in search_result.notes)}") print(f" - 总评论数：{sum(note.comments for note in search_result.notes)}") # 显示部分内容示例 print(f"\n📝 内容示例：") first_doc = list(documents.values())[0] print(f" 文档摘要：{first_doc[:100]}...") if search_result.notes: popular_note = max(search_result.notes, key=lambda x: x.likes) print(f" 热门笔记：{popular_note.title} (👍{popular_note.likes})") return { "search_result": search_result, "documents": documents } def test_different_document_types(self): """测试不同文档类型的整合效果""" print("\n" + "="*60) print("📊 测试不同文档类型的整合效果") print("="*60) doc_types = ["travel_guide", "product_info", "user_reviews"] results = {} for doc_type in doc_types: print(f"\n🔸 处理文档类型：{doc_type}") documents = self.create_mock_documents(doc_type) total_length = sum(len(content) for content in documents.values()) results[doc_type] = { "document_count": len(documents), "content_length": total_length, "files": list(documents.keys()) } print(f" ✅ 文档数量：{results[doc_type]['document_count']}") print(f" ✅ 内容长度：{results[doc_type]['content_length']}") print(f" ✅ 文件列表：{results[doc_type]['files']}") return results def test_content_filtering(self): """测试内容过滤和筛选""" print("\n" + "="*60) print("🔍 测试内容过滤和筛选") print("="*60) search_result = self.get_search_result() # 按点赞数筛选 high_quality_notes = [note for note in search_result.notes if note.likes >= 50] print(f"📈 高质量笔记（点赞≥50）：{len(high_quality_notes)} 条") # 按评论数筛选 interactive_notes = [note for note in search_result.notes if note.comments >= 10] print(f"💬 高互动笔记（评论≥10）：{len(interactive_notes)} 条") # 按标题关键词筛选 guide_notes = [note for note in search_result.notes if '攻略' in note.title or '指南' in note.title] print(f"📋 攻略类笔记：{len(guide_notes)} 条") # 显示统计信息 if search_result.notes: avg_likes = sum(note.likes for note in search_result.notes) / len(search_result.notes) avg_comments = sum(note.comments for note in search_result.notes) / len(search_result.notes) print(f"\n📊 统计信息：") print(f" 平均点赞数：{avg_likes:.1f}") print(f" 平均评论数：{avg_comments:.1f}") # 找出最受欢迎的笔记 most_liked = max(search_result.notes, key=lambda x: x.likes) most_commented = max(search_result.notes, key=lambda x: x.comments) print(f"\n🏆 最受欢迎：") print(f" 最多点赞：《{most_liked.title}》- {most_liked.likes} 赞") print(f" 最多评论：《{most_commented.title}》- {most_commented.comments} 评论") return { "total_notes": len(search_result.notes), "high_quality_notes": len(high_quality_notes), "interactive_notes": len(interactive_notes), "guide_notes": len(guide_notes) } async def test_real_ai_integration(self): """测试真实AI整合功能""" print("\n" + "="*60) print("🤖 测试真实AI整合功能") print("="*60) try: # 准备数据 search_result = self.get_search_result() documents = self.create_mock_documents("travel_guide") print("🔄 正在准备数据...") # 格式化文档内容 document_content = self._format_document_content(documents) # 格式化小红书笔记内容 xhs_content = self._format_xhs_notes(search_result.notes) # 构建prompt system_prompt = self.prompt_template.get_system_prompt() user_prompt = self.prompt_template.build_user_prompt( keywords=search_result.keyword or "馥桂萌宠园", document_content=document_content or "暂无文档内容", xhs_notes_content=xhs_content or "暂无笔记内容" ) print(f"📝 System Prompt长度: {len(system_prompt)} 字符") print(f"📝 User Prompt长度: {len(user_prompt)} 字符") # 调用真实AI print("🔄 正在调用AI进行内容整合...") start_time = datetime.now() response_text, input_tokens, output_tokens, time_cost = await self.ai_agent.generate_text( system_prompt=system_prompt, user_prompt=user_prompt, use_stream=True, stage="content_integration", ) end_time = datetime.now() print("✅ AI整合完成") print(f"📄 生成内容长度：{len(response_text)} 字符") print(f"🎯 输入Token数：{input_tokens}") print(f"🎯 输出Token数：{output_tokens}") print(f"⏱️ 处理时间：{time_cost:.2f} 秒") # 尝试解析JSON try: # 使用file_io模块的JSON处理功能 from utils.file_io import process_llm_json_text parsed_json = process_llm_json_text(response_text) if parsed_json: print("✅ JSON解析成功") print(f"📊 解析结果包含 {len(parsed_json)} 个顶级字段") # 显示解析后的结构 if isinstance(parsed_json, dict): print("🏗️ 数据结构：") for key, value in parsed_json.items(): if isinstance(value, dict): print(f" {key}: 包含 {len(value)} 个子字段") elif isinstance(value, list): print(f" {key}: 列表，包含 {len(value)} 个项目") else: print(f" {key}: {type(value).__name__}") formatted_response = json.dumps(parsed_json, ensure_ascii=False, indent=2) else: print("⚠️ JSON解析失败，使用原始响应") formatted_response = response_text except Exception as parse_error: print(f"⚠️ JSON解析出错: {parse_error}") formatted_response = response_text print(f"\n📝 AI响应预览：") preview_text = formatted_response[:500] + "..." if len(formatted_response) > 500 else formatted_response print(preview_text) return { "ai_response": formatted_response, "content_length": len(response_text), "input_tokens": input_tokens, "output_tokens": output_tokens, "processing_time": time_cost, "source_notes": len(search_result.notes), "source_docs": len(documents), "json_parsed": parsed_json is not None } except Exception as e: print(f"❌ AI整合测试失败：{e}") import traceback traceback.print_exc() return None def test_export_formats(self): """测试不同导出格式""" print("\n" + "="*60) print("📤 测试不同导出格式") print("="*60) search_result = self.get_search_result() documents = self.create_mock_documents("travel_guide") # 模拟不同格式的导出 formats = { "summary": "简要摘要格式", "blog_post": "博客文章格式", "travel_guide": "旅游攻略格式", "product_sales": "产品销售格式", "attraction_standard": "景点标准格式" } export_results = {} for format_type, description in formats.items(): print(f"🔸 生成{description}...") # 模拟不同格式的内容生成 mock_content = self._generate_format_content(format_type, search_result, documents) export_results[format_type] = { "content": mock_content, "length": len(mock_content), "format": format_type } print(f" ✅ 长度：{len(mock_content)} 字符") return export_results def _generate_format_content(self, format_type: str, search_result: SimpleSearchResult, documents: Dict[str, str]) -> str: """根据格式类型生成相应内容""" base_info = f"基于{len(search_result.notes)}篇小红书游记和官方资料" if format_type == "summary": return f"【简要摘要】{base_info}，馥桂萌宠园是嘉定区优质亲子游目的地，特色萌宠互动体验，适合周末家庭出游。" elif format_type == "blog_post": return f"""# 周末带娃新选择 | 上海馥桂萌宠园深度体验 今天想和大家分享一个超棒的亲子游目的地——位于嘉定的馥桂萌宠园！{base_info}，这里真的是遛娃神器！ ## 为什么推荐这里？ 🦘 可以近距离接触羊驼、袋鼠等萌宠 🌸 环境优美，特别是桂花季节 👨‍👩‍👧‍👦 设施完善，非常适合家庭游 详细攻略请看下文...""" elif format_type == "travel_guide": return f"""## 上海馥桂萌宠园游玩攻略 ### 基本信息 - 地址：上海市嘉定区朱桥镇 - 门票：成人80元，儿童50元 - 开放时间：9:00-17:00 ### 交通指南 地铁11号线嘉定西站转公交，或自驾直达 ### 游玩亮点 {base_info}，园区主要特色包括萌宠互动、户外游乐、休闲拍照等。 ### 注意事项 建议游玩3-4小时，春秋季节体验最佳。""" elif format_type == "product_sales": return f"""🔥【限时特惠】上海馥桂萌宠园门票 ⭐ 产品亮点： • {base_info}，口碑爆棚！ • 🦘 独特萌宠互动体验 • 🌸 网红拍照打卡圣地 • 👨‍👩‍👧‍👦 亲子游首选目的地 💰 特价套餐： 原价130元，现价只要99元！ 包含：成人门票1张 + 动物喂食包1份 📞 立即预订：xxx-xxxx-xxxx""" else: # attraction_standard return f"""景点名称：上海馥桂萌宠园 景点类型：主题公园/动物园 地理位置：上海市嘉定区朱桥镇 开放状态：正常营业 门票价格：80-50元 适宜人群：亲子家庭 推荐指数：4.2/5.0 游玩时长：3-4小时 最佳季节：春秋两季 {base_info}，具有良好的口碑和游客体验。""" def test_parameter_effects(self): """测试不同参数的效果""" print("\n" + "="*60) print("⚙️ 测试不同参数的效果") print("="*60) search_result = self.get_search_result() # 测试不同的过滤阈值 thresholds = [ {"likes": 20, "comments": 5}, {"likes": 50, "comments": 10}, {"likes": 100, "comments": 20} ] print("🔧 测试不同过滤阈值的效果：") for i, threshold in enumerate(thresholds, 1): high_likes = [note for note in search_result.notes if note.likes >= threshold["likes"]] high_comments = [note for note in search_result.notes if note.comments >= threshold["comments"]] print(f" 阈值{i}（点赞≥{threshold['likes']}, 评论≥{threshold['comments']}）:") print(f" - 符合点赞条件：{len(high_likes)} 条") print(f" - 符合评论条件：{len(high_comments)} 条") # 测试不同的内容长度限制 length_limits = [100, 200, 500] print(f"\n📏 测试不同内容长度限制的效果：") for limit in length_limits: documents = self.create_mock_documents("travel_guide") truncated_docs = {} for filename, content in documents.items(): truncated_docs[filename] = content[:limit] + "..." if len(content) > limit else content total_length = sum(len(content) for content in truncated_docs.values()) print(f" 限制{limit}字符：总长度 {total_length} 字符") return { "threshold_tests": len(thresholds), "length_tests": len(length_limits) } async def main(): """主函数""" print("🚀 启动真实AI离线整合测试...") # 检查数据文件 data_file = "batch_search_20250717_104407.json" if not os.path.exists(data_file): print(f"❌ 数据文件不存在：{data_file}") return # 初始化测试器 try: tester = OfflineIntegrationTester(data_file) except Exception as e: print(f"❌ 测试器初始化失败：{e}") import traceback traceback.print_exc() return try: # 执行各种测试 print(f"\n{'='*60}") print("🧪 开始执行真实AI离线整合测试套件") print(f"{'='*60}") # 1. 基础整合测试 basic_result = tester.test_basic_integration() # 2. 不同文档类型测试 doc_type_results = tester.test_different_document_types() # 3. 内容过滤测试 filter_results = tester.test_content_filtering() # 4. 真实AI整合测试（异步） ai_results = await tester.test_real_ai_integration() with open("ai_results.json", "w") as f: json.dump(ai_results, f, ensure_ascii=False, indent=2) # 5. 导出格式测试 export_results = tester.test_export_formats() # 6. 参数效果测试 param_results = tester.test_parameter_effects() # 输出测试总结 print(f"\n{'='*60}") print("📋 测试结果总结") print(f"{'='*60}") print(f"✅ 基础整合测试：成功") print(f"✅ 文档类型测试：处理了 {len(doc_type_results)} 种类型") print(f"✅ 内容过滤测试：识别出 {filter_results['high_quality_notes']} 条高质量笔记") if ai_results: print(f"✅ 真实AI整合测试：成功") print(f" - 处理时间：{ai_results['processing_time']:.2f}秒") print(f" - 输入Token：{ai_results['input_tokens']}") print(f" - 输出Token：{ai_results['output_tokens']}") print(f" - JSON解析：{'成功' if ai_results['json_parsed'] else '失败'}") else: print(f"❌ 真实AI整合测试：失败") # print(f"✅ 导出格式测试：生成了 {len(export_results)} 种格式") # print(f"✅ 参数效果测试：测试了 {param_results['threshold_tests']} 种阈值配置") # print(f"\n🎉 所有测试完成！") # # 提供测试建议 # print(f"\n💡 测试建议：") # print(f" 1. 可以调整AI模型参数（temperature、top_p等）来优化输出质量") # print(f" 2. 可以修改integration prompt模板来改进整合效果") # print(f" 3. 可以测试不同的文档类型组合") # print(f" 4. 可以调整内容过滤的阈值") # print(f" 5. 可以测试不同的导出格式效果") # # 显示原始数据概览 # if tester.raw_data: # summary = tester.raw_data.get('summary', {}) # print(f"\n📊 原始数据概览：") # print(f" - 总关键词数：{summary.get('total_keywords', 0)}") # print(f" - 成功搜索数：{summary.get('successful_searches', 0)}") # print(f" - 总笔记数：{summary.get('total_notes', 0)}") # print(f" - 总互动数：{summary.get('total_interactions', 0)}") except Exception as e: print(f"❌ 测试过程中发生错误：{e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())