### Python Configuration for RAG System Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This Python configuration file (config.py) sets up essential parameters for the RAG system. It defines API keys (with a recommendation to use environment variables), directory paths for data storage, and model names for embedding, reranking, and text generation. It also includes a detailed prompt template with Chain of Thought, specifying the desired JSON output format for LLM responses. An example of initializing the QwenLLM client is provided. ```python # config.py import os # API Keys (use environment variables in production) DASHSCOPE_API_KEY = os.environ.get('DASHSCOPE_API_KEY', "YOUR_API_KEY") MINERU_API_KEY = "YOUR_MINERU_API_KEY" # Directory Configuration PDF_REPORTS_DIR = "./data/raw_reports" PROCESSED_REPORTS_DIR = "./data/processed" VECTOR_STORE_DIR = "./vector_store" # Model Configuration EMBEDDING_MODEL_NAME = 'text-embedding-v2' # Alibaba Cloud embedding model RERANK_MODEL_NAME = 'gte-rerank-v2' # Reranking model GENERATION_MODEL_NAME = 'qwen-plus' # Text generation model (qwen-turbo for speed) # Prompt Template with Chain of Thought PROMPT_TEMPLATE = """ 你是一个专业的AI投资顾问，你的任务是根据提供的上下文信息，以严谨、分步的推理过程来回答用户的问题。 请严格按照以下JSON格式返回你的分析和答案，不要有任何额外的解释或说明文字。 **上下文信息:** --- {context} --- **用户问题:** {question} **你的输出必须是严格的JSON格式，如下所示:** ```json {{ "reasoning_steps": [ "第一步推理：分析问题的核心需求。", "第二步推理：在上下文中定位与问题最相关的关键信息。", "第三步推理：基于找到的信息进行逻辑推导。", "第四步推理：总结推导过程，形成最终结论。" ], "reasoning_summary": "对整个推理过程的简要概括。", "relevant_context": "直接引用上下文中最核心的一段或几段原文。", "final_answer": "根据以上分析得出的最终、直接的答案。" }} ``` """ # Usage example from core.llm_service import QwenLLM llm = QwenLLM(api_key=DASHSCOPE_API_KEY) ``` -------------------------------- ### FastAPI Application Setup with CORS and Lifespan Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt Sets up a FastAPI application for a RAG enterprise knowledge base Q&A API. It includes defining a request schema using Pydantic, configuring CORS middleware for cross-origin requests, implementing lifespan management for service initialization, and a health check endpoint. The application is configured to run using uvicorn. ```python from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import uvicorn from pydantic import BaseModel from contextlib import asynccontextmanager # Placeholder for QAService, assuming it's defined elsewhere class QAService: def __init__(self): print("Initializing QAService...") # Actual initialization logic here pass # Define request schema class AskRequest(BaseModel): query: str top_k: int = 20 rerank_top_n: int = 5 # Initialize FastAPI app with lifespan management @asynccontextmanager async def lifespan(app: FastAPI): print("应用启动... 正在初始化QA服务...") app.state.qa_service = QAService() print("QA服务初始化完成。") yield print("应用关闭。") app = FastAPI( title="RAG 企业知识库问答 API", description="基于RAG架构的企业知识库问答API服务", version="1.0.0", lifespan=lifespan ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Health check endpoint @app.get("/") async def root(): return {"message": "欢迎使用RAG企业知识库问答API！服务运行正常。"} # Start server if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True) ``` -------------------------------- ### Core RAG Service QAService.ask() Functionality (Python) Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This Python code snippet showcases the core RAG functionality using the QAService class. It initializes the service, which loads the vector database and LLM, and then calls the ask method with a query and retrieval/reranking parameters. The output includes the final answer, detailed reasoning steps, and information about the source documents used. ```python from core.qa_service import QAService # Initialize QA service (loads vector database and LLM) qa_service = QAService() # Execute complete RAG pipeline result = qa_service.ask( query="中芯国际2024年主营业务是什么？", top_k=20, # Retrieve 20 candidates rerank_top_n=5 # Rerank to top 5 ) # Access structured response print("Final Answer:", result['final_answer']) print("\nReasoning Steps:") for i, step in enumerate(result['reasoning_steps'], 1): print(f" {i}. {step}") print("\nSource Documents:") for doc in result['raw_context']: print(f" - Source: {doc['metadata']['source']}") print(f" Score: {doc['metadata']['score']}") print(f" Content: {doc['page_content'][:100]}...") ``` -------------------------------- ### Alibaba Cloud DashScope API Integration with QwenLLM (Python) Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This Python code demonstrates the integration with Alibaba Cloud's DashScope API via the QwenLLM class. It covers generating text embeddings in batches, reranking documents based on query relevance, and generating chat completions with a specified system prompt. An API key is required for initialization. ```python from core.llm_service import QwenLLM # Initialize LLM service llm = QwenLLM(api_key="your_dashscope_api_key") # Generate text embeddings for documents texts = ["阿里巴巴是中国的电商公司", "亚马逊是美国的电商公司"] embeddings = llm.get_text_embeddings_batch(texts) print(f"Generated {len(embeddings)} embeddings, dimension: {len(embeddings[0])}") # Rerank documents by relevance query = "全球最大的电商公司是哪家？" documents = [ "阿里巴巴集团是中国的跨国科技公司，专注于电子商务。", "亚马逊公司是一家位于美国西雅图的跨国电子商务企业。", "京东是中国一家自营式综合网络零售商。" ] reranked_docs = llm.get_rerank_documents(query, documents, top_n=2) print("Reranked results:", reranked_docs) # Generate chat completion prompt = "请根据以上信息回答：谁是全球最大的电商公司？" response = llm.get_chat_completion( prompt=prompt, system_prompt="你是一个专业的AI投资顾问。" ) print("LLM Response:", response) ``` -------------------------------- ### Load Database and Perform Similarity Search Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt Loads an existing knowledge base and performs a similarity search based on a given query. It then iterates through the results, printing the source and a snippet of the content for each document. This is useful for retrieving relevant information from the knowledge base. ```python db = kb_manager.load_db() results = kb_manager.similarity_search( query="中芯国际的营收情况", k=5 ) for doc in results: print(f"Source: {doc.metadata.get('source')}") print(f"Content: {doc.page_content[:150]}...\n") ``` -------------------------------- ### Execute Complete RAG Q&A Pipeline using requests (Python) Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This Python code snippet demonstrates how to send a user question to the FastAPI backend's /api/ask endpoint to execute the full RAG pipeline. It specifies the query, the number of documents to retrieve, and the number of documents after reranking. The response is a JSON object containing reasoning steps, summary, relevant context, final answer, and raw source documents. ```python import requests # Send question to RAG API response = requests.post( 'http://localhost:8000/api/ask', json={ 'query': '中芯国际的2024年第一季度营收是多少？', 'top_k': 20, # Number of documents to retrieve 'rerank_top_n': 5 # Number of documents after reranking } ) # Response structure result = response.json() # { # "reasoning_steps": [ # "第一步推理：分析问题的核心需求。", # "第二步推理：在上下文中定位与问题最相关的关键信息。", # ... # ], # "reasoning_summary": "对整个推理过程的简要概括。", # "relevant_context": "直接引用上下文中最核心的一段或几段原文", # "final_answer": "根据以上分析得出的最终、直接的答案。", # "raw_context": [...] # Array of source documents with metadata # } print(f"Answer: {result['final_answer']}") print(f"Reasoning: {result['reasoning_summary']}") ``` -------------------------------- ### Vector Database Management with KnowledgeBaseManager (Python) Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This Python code snippet illustrates the use of KnowledgeBaseManager for handling vector database operations. It includes loading JSON documents from a processed directory, splitting them into manageable chunks with overlap, and then creating and persisting the vector database using ChromaDB. Key parameters include chunk size and overlap. ```python from core.knowledge_base_manager import KnowledgeBaseManager # Initialize knowledge base manager kb_manager = KnowledgeBaseManager( processed_dir='./data/processed', persist_directory='./vector_store' ) # Load JSON documents from processed directory documents = kb_manager.load_documents() print(f"Loaded {len(documents)} document chunks") # Split documents into smaller chunks chunks = kb_manager.split_documents( documents, chunk_size=500, chunk_overlap=50 ) print(f"Split into {len(chunks)} chunks") # Create and persist vector database db = kb_manager.create_and_persist_db(chunks) print("Vector database created and persisted") ``` -------------------------------- ### React Frontend for RAG System Querying Source: https://context7.com/waymanhe/rag-company-wayman/llms.txt This React component provides a user interface for sending queries to the RAG system and displaying real-time results. It uses Ant Design for UI elements and Axios for API communication. The component handles user input, sends POST requests to the backend '/api/ask' endpoint, and displays the generated answer, reasoning steps, and source documents. Error handling and loading states are included. ```jsx import { useState } from 'react' import axios from 'axios' import { Input, Button, Card, Typography, Spin } from 'antd' const { TextArea } = Input // Configure API client const apiClient = axios.create({ baseURL: 'http://localhost:8000', }) function App() { const [query, setQuery] = useState('中芯国际的2024年主营业务是多少？') const [result, setResult] = useState(null) const [isLoading, setIsLoading] = useState(false) const [error, setError] = useState('') // Handle answer generation const handleGenerateAnswer = async () => { if (!query.trim()) { setError('请输入问题。') return } setIsLoading(true) setResult(null) setError('') try { const response = await apiClient.post('/api/ask', { query }) setResult(response.data) // Access response fields console.log('Final Answer:', response.data.final_answer) console.log('Reasoning Steps:', response.data.reasoning_steps) console.log('Source Documents:', response.data.raw_context) } catch (err) { setError('请求失败，请检查后端服务是否正在运行') console.error(err) } finally { setIsLoading(false) } } return (