Initial commit: OpenClaw workspace with habit bot and flashcard app
This commit is contained in:
joungmin
203
rag_cli.py
Executable file
203
rag_cli.py
Executable file
@@ -0,0 +1,203 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Oracle RAG CLI - Ultra lightweight RAG query tool
|
||||
Usage: python rag_cli.py "your question here"
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Try to import oracledb, use placeholder if not available
|
||||
try:
|
||||
import oracledb
|
||||
ORACLE_AVAILABLE = True
|
||||
except ImportError:
|
||||
ORACLE_AVAILABLE = False
|
||||
print("⚠️ oracledb not installed. Run: pip install oracledb")
|
||||
|
||||
# Oracle connection config (for when oracledb is available)
|
||||
ORACLE_CONFIG = {
|
||||
"user": "admin",
|
||||
"password": "Carter55@26@1",
|
||||
"dsn": "h8i4i0g8cxtd2lpf_high",
|
||||
"wallet_location": "/Users/joungmin/devkit/db_conn/Wallet_H8I4I0G8CXTD2LPF"
|
||||
}
|
||||
|
||||
|
||||
def get_connection():
|
||||
"""Get Oracle connection"""
|
||||
if not ORACLE_AVAILABLE:
|
||||
return None
|
||||
|
||||
try:
|
||||
os.environ['TNS_ADMIN'] = ORACLE_CONFIG['wallet_location']
|
||||
return oracledb.connect(
|
||||
user=ORACLE_CONFIG['user'],
|
||||
password=ORACLE_CONFIG['password'],
|
||||
dsn=ORACLE_CONFIG['dsn'],
|
||||
wallet_location=ORACLE_CONFIG['wallet_location']
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"❌ Oracle connection failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def check_rag_procedures(cursor):
|
||||
"""Check which RAG procedures exist"""
|
||||
cursor.execute("""
|
||||
SELECT object_name, object_type
|
||||
FROM user_objects
|
||||
WHERE object_name LIKE '%RAG%' OR object_name LIKE '%EMBED%'
|
||||
ORDER BY object_name
|
||||
""")
|
||||
results = cursor.fetchall()
|
||||
return results
|
||||
|
||||
|
||||
def rag_query(question: str, top_k: int = 5) -> str:
|
||||
"""Query Oracle RAG system"""
|
||||
conn = get_connection()
|
||||
if not conn:
|
||||
return "❌ No Oracle connection available"
|
||||
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
# Check available procedures
|
||||
procedures = check_rag_procedures(cursor)
|
||||
|
||||
if procedures:
|
||||
proc_names = [p[0] for p in procedures]
|
||||
print(f"📦 Found RAG procedures: {', '.join(proc_names)}")
|
||||
|
||||
# Try rag_ask if exists
|
||||
if 'RAG_ASK' in [p.upper() for p in proc_names]:
|
||||
cursor.execute("SELECT rag_ask(:1, :2) FROM DUAL", [question, top_k])
|
||||
result = cursor.fetchone()
|
||||
if result and result[0]:
|
||||
return result[0]
|
||||
else:
|
||||
print("📦 No RAG procedures found. Checking doc_chunks table...")
|
||||
|
||||
# Check if doc_chunks exists
|
||||
cursor.execute("""
|
||||
SELECT table_name FROM user_tables
|
||||
WHERE table_name LIKE '%CHUNK%' OR table_name LIKE '%DOC%'
|
||||
""")
|
||||
tables = cursor.fetchall()
|
||||
if tables:
|
||||
print(f"📦 Found tables: {', '.join([t[0] for t in tables])}")
|
||||
return vector_search_fallback(question, cursor, top_k)
|
||||
else:
|
||||
return "❌ No document tables found. Please run your ingestion pipeline first."
|
||||
|
||||
return "⚠️ RAG query returned no results"
|
||||
|
||||
except Exception as e:
|
||||
return f"❌ Query failed: {e}"
|
||||
finally:
|
||||
cursor.close()
|
||||
conn.close()
|
||||
|
||||
|
||||
def vector_search_fallback(question: str, cursor, top_k: int = 5) -> str:
|
||||
"""Direct vector search if RAG procedure not available"""
|
||||
# Check if embed_vector column exists
|
||||
try:
|
||||
cursor.execute("""
|
||||
SELECT column_name
|
||||
FROM user_tab_columns
|
||||
WHERE table_name = 'DOC_CHUNKS' AND column_name = 'EMBED_VECTOR'
|
||||
""")
|
||||
if not cursor.fetchone():
|
||||
return "⚠️ doc_chunks exists but no EMBED_VECTOR column found."
|
||||
|
||||
# Check for data
|
||||
cursor.execute("SELECT COUNT(*) FROM doc_chunks")
|
||||
count = cursor.fetchone()[0]
|
||||
if count == 0:
|
||||
return f"⚠️ doc_chunks is empty (0 rows). Ingest documents first."
|
||||
|
||||
# For now, just show status
|
||||
return f"""📊 doc_chunks status:
|
||||
- Total chunks: {count}
|
||||
- Vector search: Available (VECTOR column exists)
|
||||
- RAG procedure: Not yet created
|
||||
|
||||
To enable RAG:
|
||||
1. Create RAG procedures (see Oracle RAG Lightweight.md)
|
||||
2. Or ingest documents via your pipeline"""
|
||||
|
||||
except Exception as e:
|
||||
return f"❌ Vector search failed: {e}"
|
||||
|
||||
|
||||
def embed_text(text: str) -> str:
|
||||
"""Generate embedding using MiniMax API"""
|
||||
try:
|
||||
from openai import OpenAI
|
||||
|
||||
api_key = os.environ.get("MINIMAX_API_KEY")
|
||||
if not api_key:
|
||||
return None
|
||||
|
||||
client = OpenAI(api_key=api_key, base_url="https://api.minimax.chat/v1")
|
||||
|
||||
response = client.embeddings.create(
|
||||
model="embo-01",
|
||||
input=text
|
||||
)
|
||||
|
||||
embedding = response.data[0].embedding
|
||||
return "[" + ",".join([str(x) for x in embedding]) + "]"
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ MiniMax embedding failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
print("""
|
||||
🔮 Oracle RAG CLI v1.0
|
||||
|
||||
Usage: python rag_cli.py "your question here"
|
||||
|
||||
Options:
|
||||
-k, --top-k N Number of results (default: 5)
|
||||
-h, --help Show this help
|
||||
""")
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit(0)
|
||||
|
||||
# Parse arguments
|
||||
question = ""
|
||||
top_k = 5
|
||||
|
||||
i = 1
|
||||
while i < len(sys.argv):
|
||||
arg = sys.argv[i]
|
||||
if arg in ["-k", "--top-k"] and i + 1 < len(sys.argv):
|
||||
top_k = int(sys.argv[i + 1])
|
||||
i += 2
|
||||
elif arg in ["-h", "--help"]:
|
||||
print(__doc__)
|
||||
sys.exit(0)
|
||||
else:
|
||||
question += sys.argv[i] + " "
|
||||
i += 1
|
||||
|
||||
question = question.strip()
|
||||
|
||||
if not question:
|
||||
print("❌ Please provide a question")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"\n🔍 Querying Oracle RAG: \"{question[:50]}{'...' if len(question) > 50 else ''}\"\n")
|
||||
|
||||
result = rag_query(question, top_k)
|
||||
print(result)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user