""" 高级使用示例 - 展示更多功能 """ import os import sys from pathlib import Path # 添加项目根目录到Python路径 project_root = Path(__file__).parent.parent sys.path.insert(0, str(project_root)) from src.core.llm_manager import llm_manager from src.core.vector_store import vector_store_manager from src.core.document_processor import document_processor from src.services.chat_service import chat_service from src.services.qa_service import qa_service from src.services.document_service import document_service from src.utils.logger import setup_logger from src.utils.cache import cache_manager from loguru import logger def demo_multi_provider_llm(): """演示多提供商LLM使用""" print("=== 多提供商LLM演示 ===") providers = ["openrouter", "openai", "anthropic"] for provider in providers: try: print(f"\n使用 {provider} 提供商:") # 获取可用模型 models = llm_manager.get_available_models(provider) print(f" 可用模型: {models[:3]}...") # 只显示前3个 # 测试连接 if llm_manager.test_connection(provider): print(f" {provider} 连接正常") # 创建LLM实例 llm = llm_manager.get_or_create_llm(provider) # 简单测试 response = llm.invoke("请用一句话介绍自己") print(f" 回复: {response.content[:100]}...") else: print(f" {provider} 连接失败") except Exception as e: print(f" {provider} 测试失败: {e}") def demo_advanced_chat(): """演示高级聊天功能""" print("\n=== 高级聊天功能演示 ===") # 创建多个会话 sessions = [ ("session_1", "你是一个技术专家，专门回答编程问题"), ("session_2", "你是一个创意写作助手，帮助用户创作故事"), ("session_3", "你是一个数据分析师，帮助用户分析数据") ] for session_id, system_prompt in sessions: try: # 创建会话 session = chat_service.create_session(session_id, system_prompt) print(f"\n创建会话: {session_id}") # 发送消息 if session_id == "session_1": message = "请解释什么是Python装饰器" elif session_id == "session_2": message = "请写一个关于太空探险的短故事开头" else: message = "如何分析销售数据的趋势？" response = chat_service.send_message(session_id, message) print(f" 问题: {message}") print(f" 回答: {response['assistant_message'][:150]}...") except Exception as e: print(f" 会话 {session_id} 失败: {e}") # 获取会话摘要 for session_id, _ in sessions: try: summary = chat_service.get_conversation_summary(session_id) print(f"\n会话 {session_id} 摘要: {summary[:100]}...") except Exception as e: print(f"获取会话 {session_id} 摘要失败: {e}") def demo_document_processing(): """演示文档处理功能""" print("\n=== 文档处理功能演示 ===") # 创建不同类型的测试文档 test_docs = { "technical.txt": """ 人工智能技术发展迅速，机器学习算法在各个领域都有广泛应用。 深度学习模型如GPT、BERT等在自然语言处理任务中表现出色。 计算机视觉技术也在图像识别、目标检测等方面取得了重大突破。 """, "business.txt": """ 企业数字化转型是当前的重要趋势，云计算、大数据分析等技术正在改变商业模式。 远程办公和在线协作工具的使用率大幅提升，推动了企业效率的提升。 数据安全和隐私保护成为企业关注的重点问题。 """, "creative.txt": """ 创意写作需要丰富的想象力和独特的视角，每个故事都有其独特的魅力。 文学创作不仅仅是文字的堆砌，更是情感和思想的表达。 好的作品能够引起读者的共鸣，传递深刻的人生哲理。 """ } # 创建文档 for filename, content in test_docs.items(): with open(filename, "w", encoding="utf-8") as f: f.write(content.strip()) try: # 批量处理文档 for filename in test_docs.keys(): print(f"\n处理文档: {filename}") # 验证文档 validation = document_service.validate_document(filename) print(f" 验证结果: {validation['valid']}") if validation["valid"]: # 处理文档 result = document_service.process_and_store_document( filename, collection_name=f"demo_{filename.split('.')[0]}", additional_metadata={"category": "demo", "filename": filename} ) print(f" 处理完成: {result['processed_documents']} 个片段") # 搜索文档 search_queries = ["人工智能", "企业", "创意"] for query in search_queries: print(f"\n搜索关键词: {query}") results = document_service.search_documents_with_scores(query, k=3) for i, result in enumerate(results, 1): print(f" 结果 {i}: {result['content'][:100]}... (相似度: {result['similarity_score']:.3f})") # 获取集合信息 collections = document_service.get_document_collections() print(f"\n文档集合: {collections}") except Exception as e: print(f"文档处理演示失败: {e}") # 清理测试文件 for filename in test_docs.keys(): if os.path.exists(filename): os.remove(filename) def demo_advanced_qa(): """演示高级问答功能""" print("\n=== 高级问答功能演示 ===") # 创建知识库文档 knowledge_doc = """ LangChain是一个用于开发由语言模型驱动的应用程序的框架。 它提供了以下主要功能： 1. 模型管理：支持多种LLM提供商，如OpenAI、Anthropic等 2. 提示管理：提供灵活的提示模板和链式处理 3. 记忆管理：支持对话历史和上下文管理 4. 索引和检索：集成向量数据库，支持文档检索 5. 代理和工具：支持工具调用和自动化任务 LangChain的主要优势： - 模块化设计，易于扩展 - 丰富的集成选项 - 活跃的社区支持 - 详细的文档和示例 """ # 创建测试文档 test_file = "langchain_knowledge.txt" with open(test_file, "w", encoding="utf-8") as f: f.write(knowledge_doc) try: # 处理文档并添加到问答系统 documents = document_processor.process_document_pipeline(test_file) qa_service.add_documents_for_qa(documents, collection_name="langchain_qa") # 批量问答 questions = [ "LangChain是什么？", "LangChain有哪些主要功能？", "LangChain的优势是什么？", "如何使用LangChain进行文档检索？" ] print("批量问答测试:") results = qa_service.batch_qa(questions, collection_name="langchain_qa") for i, result in enumerate(results, 1): print(f"\n问题 {i}: {result['question']}") print(f"答案: {result['answer'][:200]}...") print(f"来源数量: {result['source_count']}") # 相似文档搜索 print(f"\n相似文档搜索:") similar_docs = qa_service.search_similar_documents( "LangChain框架", k=3, collection_name="langchain_qa" ) for i, doc in enumerate(similar_docs, 1): print(f" 文档 {i}: {doc.page_content[:150]}...") except Exception as e: print(f"高级问答演示失败: {e}") # 清理测试文件 if os.path.exists(test_file): os.remove(test_file) def demo_cache_management(): """演示缓存管理功能""" print("\n=== 缓存管理功能演示 ===") # 测试缓存功能 test_data = {"key": "value", "number": 42, "list": [1, 2, 3]} # 设置缓存 cache_manager.set("test_key", test_data, ttl=60) print("设置缓存: test_key") # 获取缓存 cached_data = cache_manager.get("test_key") if cached_data: print(f"获取缓存成功: {cached_data}") else: print("缓存获取失败") # 删除缓存 cache_manager.delete("test_key") print("删除缓存: test_key") # 验证删除 cached_data = cache_manager.get("test_key") if cached_data is None: print("缓存删除成功") else: print("缓存删除失败") def main(): """主函数""" print("AI Architecture 高级使用示例") print("=" * 60) # 设置日志 setup_logger() # 检查环境变量 if not os.getenv("OPENROUTER_API_KEY"): print("警告: 未设置OPENROUTER_API_KEY环境变量") print("请复制env.example为.env并设置您的OpenRouter API密钥") return # 运行演示 demo_multi_provider_llm() demo_advanced_chat() demo_document_processing() demo_advanced_qa() demo_cache_management() print("\n高级示例运行完成！") if __name__ == "__main__": main()