feat: async background chat, performance scaling, and dead code cleanup
Async chat processing eliminates 504 timeouts - POST returns immediately, backend processes in background, frontend polls for results with timestamps. Scale to ~12 simultaneous chats via 8 uvicorn workers + 16-thread executor. Parallelized MCP data collection, 2h session cache, 5min tool timeout. Full dead code cleanup across backend, frontend, MCP, docker, and nginx.
This commit is contained in:
nogueiraguh
421
backend/app.py
421
backend/app.py
@@ -5,11 +5,12 @@ OCI Account Explorer, MCP Server registry with VectorDB tool integration,
|
||||
Autonomous DB vector storage, CIS reports, chat agent, audit log.
|
||||
"""
|
||||
import os, json, uuid, hashlib, hmac, time, base64, struct, secrets, subprocess
|
||||
import shutil, asyncio, sqlite3, logging, socket, re
|
||||
import shutil, asyncio, sqlite3, logging, re, concurrent.futures
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Optional, List, Dict, Any
|
||||
from contextlib import contextmanager
|
||||
from functools import partial
|
||||
|
||||
from fastapi import (
|
||||
FastAPI, HTTPException, Depends, Request, UploadFile, File, Form,
|
||||
@@ -37,6 +38,7 @@ for d in [DATA, OCI_DIR, REPORTS, MCP_DIR, WALLET_DIR]:
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
_running_reports: dict[str, asyncio.subprocess.Process] = {} # rid → subprocess
|
||||
_chat_executor = concurrent.futures.ThreadPoolExecutor(max_workers=16, thread_name_prefix="chat")
|
||||
|
||||
# ── Chat Memory Compaction Settings ──
|
||||
COMPACT_TOKEN_THRESHOLD = 8000 # estimated tokens before triggering compaction
|
||||
@@ -199,9 +201,10 @@ OCI_REGIONS = {
|
||||
# ── Database ──────────────────────────────────────────────────────────────────
|
||||
@contextmanager
|
||||
def db():
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn = sqlite3.connect(str(DB_PATH), timeout=30, check_same_thread=False)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA busy_timeout=30000")
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
try:
|
||||
yield conn
|
||||
@@ -379,6 +382,11 @@ def init_db():
|
||||
c.execute(f"ALTER TABLE reports ADD COLUMN {col}")
|
||||
except sqlite3.OperationalError:
|
||||
pass
|
||||
for col in ["status TEXT DEFAULT 'done'"]:
|
||||
try:
|
||||
c.execute(f"ALTER TABLE chat_messages ADD COLUMN {col}")
|
||||
except sqlite3.OperationalError:
|
||||
pass
|
||||
# Migrate legacy table_name from adb_vector_configs into adb_vector_tables
|
||||
try:
|
||||
for cfg_row in c.execute("SELECT id, table_name FROM adb_vector_configs WHERE table_name IS NOT NULL AND table_name != ''").fetchall():
|
||||
@@ -496,10 +504,6 @@ class GenAIConfigReq(BaseModel):
|
||||
temperature: float = 1; max_tokens: int = 6000; top_p: float = 0.95
|
||||
top_k: int = 1; frequency_penalty: float = 0; presence_penalty: float = 0
|
||||
is_default: bool = False
|
||||
class ADBVectorReq(BaseModel):
|
||||
config_name: str; dsn: str; username: str; password: str
|
||||
wallet_password: Optional[str] = None; table_name: str = "CIS_EMBEDDINGS"; use_mtls: bool = True
|
||||
genai_config_id: Optional[str] = None; embedding_model_id: str = "cohere.embed-v4.0"
|
||||
class IngestDocReq(BaseModel):
|
||||
adb_config_id: str; documents: List[Dict[str, Any]]; table_name: Optional[str] = None
|
||||
class MCPServerReq(BaseModel):
|
||||
@@ -1035,7 +1039,7 @@ def _call_genai(gc: dict, message: str, history: list = None, tools: list = None
|
||||
config=config,
|
||||
service_endpoint=endpoint,
|
||||
retry_strategy=oci.retry.NoneRetryStrategy(),
|
||||
timeout=(10, 240)
|
||||
timeout=(10, 600)
|
||||
)
|
||||
|
||||
# System prompt
|
||||
@@ -1733,8 +1737,7 @@ async def update_adb(
|
||||
_config_log("adb", vid, config_name, "success", "save", f"Conexão atualizada: {config_name} ({dsn})", u["id"], u["username"])
|
||||
return {"id": vid, "config_name": config_name}
|
||||
|
||||
import re as _re
|
||||
_TABLE_NAME_RE = _re.compile(r'^[A-Z][A-Z0-9_]{0,127}$')
|
||||
_TABLE_NAME_RE = re.compile(r'^[A-Z][A-Z0-9_]{0,127}$')
|
||||
|
||||
@app.get("/api/adb/{vid}/tables")
|
||||
async def list_adb_tables(vid: str, u=Depends(current_user)):
|
||||
@@ -1917,12 +1920,15 @@ def _get_adb_and_genai(vid: str):
|
||||
async def preview_report_chunks(rid: str, u=Depends(current_user)):
|
||||
"""Preview the chunks that will be generated from a CIS report before embedding."""
|
||||
with db() as c:
|
||||
r = c.execute("SELECT report_data,tenancy_name FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
r = c.execute("SELECT json_path,tenancy_name FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
if not r: raise HTTPException(404, "Report not found or not completed")
|
||||
report_data = r["report_data"]
|
||||
if isinstance(report_data, str):
|
||||
try: report_data = json.loads(report_data)
|
||||
except: raise HTTPException(400, "Invalid report data")
|
||||
json_path = r["json_path"]
|
||||
if not json_path or not Path(json_path).exists():
|
||||
raise HTTPException(400, "Report JSON file not found")
|
||||
try:
|
||||
report_data = json.loads(Path(json_path).read_text())
|
||||
except Exception:
|
||||
raise HTTPException(400, "Invalid report data")
|
||||
documents = _chunk_report_by_section(report_data)
|
||||
return {"tenancy": report_data.get("tenancy", "unknown"),
|
||||
"regions": report_data.get("regions", []),
|
||||
@@ -1935,12 +1941,15 @@ async def embed_report(rid: str, req: dict, bg: BackgroundTasks, u=Depends(requi
|
||||
vid = req.get("adb_config_id")
|
||||
if not vid: raise HTTPException(400, "adb_config_id is required")
|
||||
with db() as c:
|
||||
r = c.execute("SELECT report_data,tenancy_name FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
r = c.execute("SELECT json_path,tenancy_name FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
if not r: raise HTTPException(404, "Report not found or not completed")
|
||||
report_data = r["report_data"]
|
||||
if isinstance(report_data, str):
|
||||
try: report_data = json.loads(report_data)
|
||||
except: raise HTTPException(400, "Invalid report data")
|
||||
json_path = r["json_path"]
|
||||
if not json_path or not Path(json_path).exists():
|
||||
raise HTTPException(400, "Report JSON file not found")
|
||||
try:
|
||||
report_data = json.loads(Path(json_path).read_text())
|
||||
except Exception:
|
||||
raise HTTPException(400, "Invalid report data")
|
||||
documents = _chunk_report_by_section(report_data)
|
||||
if not documents: raise HTTPException(400, "No sections found in report")
|
||||
cfg, gc = _get_adb_and_genai(vid)
|
||||
@@ -2150,10 +2159,7 @@ async def get_report(rid, u=Depends(current_user)):
|
||||
with db() as c: r=c.execute("SELECT * FROM reports WHERE id=?",(rid,)).fetchone()
|
||||
if not r: raise HTTPException(404)
|
||||
if u["role"]!="admin" and r["user_id"]!=u["id"]: raise HTTPException(403)
|
||||
d=dict(r)
|
||||
d.pop("report_data", None)
|
||||
d.pop("mcp_server_id", None)
|
||||
return d
|
||||
return dict(r)
|
||||
|
||||
@app.get("/api/reports/{rid}/files")
|
||||
async def list_report_files(rid: str, u=Depends(current_user)):
|
||||
@@ -2196,12 +2202,13 @@ async def report_dl(rid, fmt: str = Query("json"), u=Depends(current_user)):
|
||||
# ── Chat Agent ────────────────────────────────────────────────────────────────
|
||||
# (endpoints defined below _chat_core, after _agent_respond)
|
||||
|
||||
async def _chat_core_impl(msg: ChatMsg, u, attachments: list = None):
|
||||
"""Internal chat implementation — called by both /api/chat and /api/chat/upload."""
|
||||
def _chat_start(msg: ChatMsg, u, attachments: list = None):
|
||||
"""Start a chat: save user msg, resolve config, return (sid, mid, genai_cfg) or immediate response.
|
||||
If genai_cfg is None, returns immediate fallback response in mid field as dict."""
|
||||
sid = msg.session_id or str(uuid.uuid4())
|
||||
with db() as c:
|
||||
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content,model_id) VALUES (?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), sid, u["id"], "user", msg.message, None))
|
||||
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content,model_id,status) VALUES (?,?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), sid, u["id"], "user", msg.message, None, "done"))
|
||||
|
||||
genai_cfg = None
|
||||
if msg.genai_config_id:
|
||||
@@ -2209,7 +2216,6 @@ async def _chat_core_impl(msg: ChatMsg, u, attachments: list = None):
|
||||
row = c.execute("SELECT * FROM genai_configs WHERE id=?", (msg.genai_config_id,)).fetchone()
|
||||
if row:
|
||||
genai_cfg = dict(row)
|
||||
# Override params from inline chat settings if provided
|
||||
if msg.temperature is not None: genai_cfg["temperature"] = msg.temperature
|
||||
if msg.max_tokens is not None: genai_cfg["max_tokens"] = msg.max_tokens
|
||||
if msg.top_p is not None: genai_cfg["top_p"] = msg.top_p
|
||||
@@ -2217,7 +2223,6 @@ async def _chat_core_impl(msg: ChatMsg, u, attachments: list = None):
|
||||
if msg.frequency_penalty is not None: genai_cfg["frequency_penalty"] = msg.frequency_penalty
|
||||
if msg.presence_penalty is not None: genai_cfg["presence_penalty"] = msg.presence_penalty
|
||||
elif msg.model_id and msg.oci_config_id:
|
||||
# Direct model mode: build synthetic config dict
|
||||
with db() as c:
|
||||
oci_row = c.execute("SELECT * FROM oci_configs WHERE id=?", (msg.oci_config_id,)).fetchone()
|
||||
if not oci_row:
|
||||
@@ -2243,148 +2248,158 @@ async def _chat_core_impl(msg: ChatMsg, u, attachments: list = None):
|
||||
"presence_penalty": msg.presence_penalty if msg.presence_penalty is not None else 0.0,
|
||||
}
|
||||
|
||||
if genai_cfg:
|
||||
try:
|
||||
history = []
|
||||
with db() as c:
|
||||
prev = c.execute("SELECT role,content FROM chat_messages WHERE session_id=? AND role IN ('user','assistant') ORDER BY created_at ASC", (sid,)).fetchall()
|
||||
history = [{"role":r["role"],"content":r["content"]} for r in prev]
|
||||
|
||||
# ── RAG: augment with vector context from ALL active ADB configs ──
|
||||
rag_context = ""
|
||||
adb_cfgs = _get_active_adb_configs(u["id"])
|
||||
if adb_cfgs:
|
||||
all_documents = []
|
||||
for adb_cfg in adb_cfgs:
|
||||
try:
|
||||
with db() as c:
|
||||
emb_genai = c.execute("SELECT * FROM genai_configs WHERE id=?", (adb_cfg["genai_config_id"],)).fetchone()
|
||||
if emb_genai:
|
||||
emb_model = adb_cfg.get("embedding_model_id", "cohere.embed-v4.0")
|
||||
query_embedding = _embed_text(msg.message, dict(emb_genai), emb_model)
|
||||
tables = _get_tables_for_config(adb_cfg["id"], active_only=True)
|
||||
if not tables:
|
||||
tables = [{"table_name": adb_cfg.get("table_name", "CIS_EMBEDDINGS")}]
|
||||
for tbl in tables:
|
||||
try:
|
||||
documents = _vector_search(adb_cfg, query_embedding, top_k=5, table_name=tbl["table_name"])
|
||||
if documents:
|
||||
for doc in documents:
|
||||
doc["source"] = f"{doc.get('source', 'unknown')} [{tbl['table_name']}]"
|
||||
all_documents.extend(documents)
|
||||
log.info(f"RAG: Retrieved {len(documents)} docs from {tbl['table_name']}")
|
||||
except Exception as te:
|
||||
log.warning(f"RAG search failed for table {tbl['table_name']}: {te}")
|
||||
except Exception as e:
|
||||
log.warning(f"RAG retrieval failed for {adb_cfg.get('config_name','?')} (non-fatal): {e}")
|
||||
if all_documents:
|
||||
all_documents.sort(key=lambda d: d["distance"])
|
||||
rag_context = _build_rag_context(all_documents[:10])
|
||||
|
||||
cfg_dict = dict(genai_cfg)
|
||||
# Global system prompt from system_prompts table
|
||||
with db() as c:
|
||||
sp_row = c.execute("SELECT content FROM system_prompts WHERE agent='chat' AND is_active=1 LIMIT 1").fetchone()
|
||||
global_prompt = sp_row["content"] if sp_row and sp_row["content"] else ""
|
||||
# If RAG context found, wrap user message with context
|
||||
if rag_context:
|
||||
augmented_message = RAG_CONTEXT_TEMPLATE.format(context=rag_context, question=msg.message)
|
||||
cfg_dict["system_prompt"] = global_prompt or RAG_DEFAULT_SYSTEM_PROMPT
|
||||
else:
|
||||
augmented_message = msg.message
|
||||
if global_prompt:
|
||||
cfg_dict["system_prompt"] = global_prompt
|
||||
# Collect MCP tools if enabled
|
||||
mcp_tools = []
|
||||
tool_defs = None
|
||||
if msg.use_tools:
|
||||
mcp_tools = _get_active_mcp_tools(u["id"])
|
||||
if mcp_tools:
|
||||
tool_defs = [t["tool"] for t in mcp_tools]
|
||||
log.info(f"Chat with {len(tool_defs)} MCP tools available")
|
||||
|
||||
# ── Memory compaction: summarize old messages if history is too long ──
|
||||
if history and _should_compact(history):
|
||||
log.info(f"Compaction triggered: {len(history)} msgs, ~{_estimate_history_tokens(history)} est tokens")
|
||||
history = _compact_history(sid, u["id"], cfg_dict, history)
|
||||
log.info(f"Post-compaction: {len(history)} msgs, ~{_estimate_history_tokens(history)} est tokens")
|
||||
|
||||
hist = history[:-1] if len(history) > 1 else None
|
||||
resp_text, tool_calls, tool_calls_raw = _call_genai(cfg_dict, augmented_message, hist, tools=tool_defs, attachments=attachments)
|
||||
|
||||
# Tool use loop (max 5 iterations)
|
||||
# Accumulate extra_messages so the model sees the full tool use conversation
|
||||
all_tool_results = []
|
||||
accumulated_msgs = [] # raw OCI SDK message objects for Generic format
|
||||
iterations = 0
|
||||
api_format = GENAI_MODELS.get(genai_cfg.get("model_id", ""), {}).get("api_format", "GENERIC")
|
||||
while tool_calls and iterations < 5:
|
||||
iterations += 1
|
||||
log.info(f"Tool use iteration {iterations}: {len(tool_calls)} tool call(s)")
|
||||
iteration_results = []
|
||||
for tc in tool_calls:
|
||||
mcp_match = next((m for m in mcp_tools if m["tool"]["name"] == tc["name"]), None)
|
||||
if mcp_match:
|
||||
try:
|
||||
result = await _execute_mcp_tool(mcp_match["server"], tc["name"], tc["arguments"])
|
||||
log.info(f"Tool {tc['name']} executed successfully ({len(result)} chars)")
|
||||
except Exception as te:
|
||||
result = f"Erro ao executar tool {tc['name']}: {str(te)[:300]}"
|
||||
log.warning(f"Tool {tc['name']} failed: {te}")
|
||||
else:
|
||||
result = f"Tool {tc['name']} não encontrada nos MCP servers ativos"
|
||||
iteration_results.append({"tool_call_id": tc["id"], "name": tc["name"], "content": result})
|
||||
all_tool_results.extend(iteration_results)
|
||||
|
||||
# Build tool results in the appropriate format and call again
|
||||
if api_format == "COHERE":
|
||||
import oci
|
||||
cohere_results = []
|
||||
for tr in iteration_results:
|
||||
tc_obj = oci.generative_ai_inference.models.CohereToolCall()
|
||||
tc_obj.name = tr["name"]
|
||||
tc_obj.parameters = {}
|
||||
tr_obj = oci.generative_ai_inference.models.CohereToolResult()
|
||||
tr_obj.call = tc_obj
|
||||
tr_obj.outputs = [{"result": tr["content"]}]
|
||||
cohere_results.append(tr_obj)
|
||||
resp_text, tool_calls, tool_calls_raw = _call_genai(
|
||||
cfg_dict, augmented_message, hist, tools=tool_defs,
|
||||
tool_results_cohere=cohere_results)
|
||||
else:
|
||||
import oci
|
||||
# Accumulate: add assistant message with tool_calls + tool result messages
|
||||
assistant_msg = oci.generative_ai_inference.models.AssistantMessage()
|
||||
assistant_msg.tool_calls = tool_calls_raw
|
||||
accumulated_msgs.append(assistant_msg)
|
||||
for tr in iteration_results:
|
||||
tool_msg = oci.generative_ai_inference.models.ToolMessage()
|
||||
tool_msg.tool_call_id = tr["tool_call_id"]
|
||||
tool_content = oci.generative_ai_inference.models.TextContent()
|
||||
tool_content.text = tr["content"]
|
||||
tool_msg.content = [tool_content]
|
||||
accumulated_msgs.append(tool_msg)
|
||||
|
||||
resp_text, tool_calls, tool_calls_raw = _call_genai(
|
||||
cfg_dict, augmented_message, hist, tools=tool_defs,
|
||||
extra_messages=accumulated_msgs)
|
||||
|
||||
resp = resp_text
|
||||
except Exception as e:
|
||||
resp = f"❌ Erro GenAI: {str(e)[:400]}"
|
||||
all_tool_results = []
|
||||
else:
|
||||
if not genai_cfg:
|
||||
# No GenAI config — return immediate fallback
|
||||
resp = _agent_respond(msg.message, u)
|
||||
all_tool_results = []
|
||||
with db() as c:
|
||||
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content,model_id,status) VALUES (?,?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), sid, u["id"], "assistant", resp, None, "done"))
|
||||
return sid, {"session_id": sid, "response": resp, "model_id": None, "status": "done"}, None
|
||||
|
||||
mid = genai_cfg["model_id"] if genai_cfg else None
|
||||
# Create placeholder assistant message for background processing
|
||||
mid = str(uuid.uuid4())
|
||||
with db() as c:
|
||||
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content,model_id) VALUES (?,?,?,?,?,?)",
|
||||
(str(uuid.uuid4()), sid, u["id"], "assistant", resp, mid))
|
||||
result = {"session_id": sid, "response": resp, "model_id": mid}
|
||||
if all_tool_results:
|
||||
result["tools_used"] = [{"name": tr["name"], "result_preview": tr["content"][:200]} for tr in all_tool_results]
|
||||
return result
|
||||
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content,model_id,status) VALUES (?,?,?,?,?,?,?)",
|
||||
(mid, sid, u["id"], "assistant", "", genai_cfg.get("model_id"), "processing"))
|
||||
return sid, mid, genai_cfg
|
||||
|
||||
|
||||
async def _chat_background(mid: str, sid: str, msg: ChatMsg, user: dict, genai_cfg: dict, attachments: list = None):
|
||||
"""Background worker — processes GenAI chat, updates DB when done."""
|
||||
log.info(f"Chat background started: mid={mid}, sid={sid}")
|
||||
try:
|
||||
history = []
|
||||
with db() as c:
|
||||
prev = c.execute("SELECT role,content FROM chat_messages WHERE session_id=? AND role IN ('user','assistant') AND status='done' ORDER BY created_at ASC", (sid,)).fetchall()
|
||||
history = [{"role":r["role"],"content":r["content"]} for r in prev]
|
||||
|
||||
# ── RAG: augment with vector context from ALL active ADB configs ──
|
||||
rag_context = ""
|
||||
adb_cfgs = _get_active_adb_configs(user["id"])
|
||||
if adb_cfgs:
|
||||
all_documents = []
|
||||
for adb_cfg in adb_cfgs:
|
||||
try:
|
||||
with db() as c:
|
||||
emb_genai = c.execute("SELECT * FROM genai_configs WHERE id=?", (adb_cfg["genai_config_id"],)).fetchone()
|
||||
if emb_genai:
|
||||
emb_model = adb_cfg.get("embedding_model_id", "cohere.embed-v4.0")
|
||||
query_embedding = _embed_text(msg.message, dict(emb_genai), emb_model)
|
||||
tables = _get_tables_for_config(adb_cfg["id"], active_only=True)
|
||||
if not tables:
|
||||
tables = [{"table_name": adb_cfg.get("table_name", "CIS_EMBEDDINGS")}]
|
||||
for tbl in tables:
|
||||
try:
|
||||
documents = _vector_search(adb_cfg, query_embedding, top_k=5, table_name=tbl["table_name"])
|
||||
if documents:
|
||||
for doc in documents:
|
||||
doc["source"] = f"{doc.get('source', 'unknown')} [{tbl['table_name']}]"
|
||||
all_documents.extend(documents)
|
||||
log.info(f"RAG: Retrieved {len(documents)} docs from {tbl['table_name']}")
|
||||
except Exception as te:
|
||||
log.warning(f"RAG search failed for table {tbl['table_name']}: {te}")
|
||||
except Exception as e:
|
||||
log.warning(f"RAG retrieval failed for {adb_cfg.get('config_name','?')} (non-fatal): {e}")
|
||||
if all_documents:
|
||||
all_documents.sort(key=lambda d: d["distance"])
|
||||
rag_context = _build_rag_context(all_documents[:10])
|
||||
|
||||
cfg_dict = dict(genai_cfg)
|
||||
with db() as c:
|
||||
sp_row = c.execute("SELECT content FROM system_prompts WHERE agent='chat' AND is_active=1 LIMIT 1").fetchone()
|
||||
global_prompt = sp_row["content"] if sp_row and sp_row["content"] else ""
|
||||
if rag_context:
|
||||
augmented_message = RAG_CONTEXT_TEMPLATE.format(context=rag_context, question=msg.message)
|
||||
cfg_dict["system_prompt"] = global_prompt or RAG_DEFAULT_SYSTEM_PROMPT
|
||||
else:
|
||||
augmented_message = msg.message
|
||||
if global_prompt:
|
||||
cfg_dict["system_prompt"] = global_prompt
|
||||
|
||||
mcp_tools = []
|
||||
tool_defs = None
|
||||
if msg.use_tools:
|
||||
mcp_tools = _get_active_mcp_tools(user["id"])
|
||||
if mcp_tools:
|
||||
tool_defs = [t["tool"] for t in mcp_tools]
|
||||
log.info(f"Chat with {len(tool_defs)} MCP tools available")
|
||||
|
||||
if history and _should_compact(history):
|
||||
log.info(f"Compaction triggered: {len(history)} msgs, ~{_estimate_history_tokens(history)} est tokens")
|
||||
history = _compact_history(sid, user["id"], cfg_dict, history)
|
||||
log.info(f"Post-compaction: {len(history)} msgs, ~{_estimate_history_tokens(history)} est tokens")
|
||||
|
||||
hist = history[:-1] if len(history) > 1 else None
|
||||
loop = asyncio.get_event_loop()
|
||||
resp_text, tool_calls, tool_calls_raw = await loop.run_in_executor(
|
||||
_chat_executor, partial(_call_genai, cfg_dict, augmented_message, hist,
|
||||
tool_defs, None, None, attachments))
|
||||
|
||||
all_tool_results = []
|
||||
accumulated_msgs = []
|
||||
iterations = 0
|
||||
api_format = GENAI_MODELS.get(genai_cfg.get("model_id", ""), {}).get("api_format", "GENERIC")
|
||||
while tool_calls and iterations < 5:
|
||||
iterations += 1
|
||||
log.info(f"Tool use iteration {iterations}: {len(tool_calls)} tool call(s)")
|
||||
iteration_results = []
|
||||
for tc in tool_calls:
|
||||
mcp_match = next((m for m in mcp_tools if m["tool"]["name"] == tc["name"]), None)
|
||||
if mcp_match:
|
||||
try:
|
||||
result = await _execute_mcp_tool(mcp_match["server"], tc["name"], tc["arguments"])
|
||||
log.info(f"Tool {tc['name']} executed successfully ({len(result)} chars)")
|
||||
except Exception as te:
|
||||
result = f"Erro ao executar tool {tc['name']}: {str(te)[:300]}"
|
||||
log.warning(f"Tool {tc['name']} failed: {te}")
|
||||
else:
|
||||
result = f"Tool {tc['name']} não encontrada nos MCP servers ativos"
|
||||
iteration_results.append({"tool_call_id": tc["id"], "name": tc["name"], "content": result})
|
||||
all_tool_results.extend(iteration_results)
|
||||
|
||||
if api_format == "COHERE":
|
||||
import oci
|
||||
cohere_results = []
|
||||
for tr in iteration_results:
|
||||
tc_obj = oci.generative_ai_inference.models.CohereToolCall()
|
||||
tc_obj.name = tr["name"]
|
||||
tc_obj.parameters = {}
|
||||
tr_obj = oci.generative_ai_inference.models.CohereToolResult()
|
||||
tr_obj.call = tc_obj
|
||||
tr_obj.outputs = [{"result": tr["content"]}]
|
||||
cohere_results.append(tr_obj)
|
||||
resp_text, tool_calls, tool_calls_raw = await loop.run_in_executor(
|
||||
_chat_executor, partial(_call_genai, cfg_dict, augmented_message, hist,
|
||||
tool_defs, cohere_results))
|
||||
else:
|
||||
import oci
|
||||
assistant_msg = oci.generative_ai_inference.models.AssistantMessage()
|
||||
assistant_msg.tool_calls = tool_calls_raw
|
||||
accumulated_msgs.append(assistant_msg)
|
||||
for tr in iteration_results:
|
||||
tool_msg = oci.generative_ai_inference.models.ToolMessage()
|
||||
tool_msg.tool_call_id = tr["tool_call_id"]
|
||||
tool_content = oci.generative_ai_inference.models.TextContent()
|
||||
tool_content.text = tr["content"]
|
||||
tool_msg.content = [tool_content]
|
||||
accumulated_msgs.append(tool_msg)
|
||||
resp_text, tool_calls, tool_calls_raw = await loop.run_in_executor(
|
||||
_chat_executor, partial(_call_genai, cfg_dict, augmented_message, hist,
|
||||
tool_defs, None, accumulated_msgs))
|
||||
|
||||
resp = resp_text
|
||||
if all_tool_results:
|
||||
tools_info = '\n\n🔧 **Tools utilizadas:** ' + ', '.join(tr["name"] for tr in all_tool_results)
|
||||
resp += tools_info
|
||||
|
||||
with db() as c:
|
||||
c.execute("UPDATE chat_messages SET content=?, status='done' WHERE id=?", (resp, mid))
|
||||
log.info(f"Chat {mid} completed successfully")
|
||||
except Exception as e:
|
||||
log.error(f"Chat {mid} failed: {e}")
|
||||
with db() as c:
|
||||
c.execute("UPDATE chat_messages SET content=?, status='failed' WHERE id=?",
|
||||
(f"❌ Erro GenAI: {str(e)[:400]}", mid))
|
||||
|
||||
MAX_UPLOAD_FILES = 5
|
||||
MAX_UPLOAD_SIZE = 10 * 1024 * 1024 # 10MB
|
||||
@@ -2394,6 +2409,7 @@ DOC_MIMES = {"application/pdf"}
|
||||
|
||||
@app.post("/api/chat/upload")
|
||||
async def chat_with_files(
|
||||
bg: BackgroundTasks,
|
||||
message: str = Form(""),
|
||||
session_id: Optional[str] = Form(None),
|
||||
genai_config_id: Optional[str] = Form(None),
|
||||
@@ -2414,7 +2430,6 @@ async def chat_with_files(
|
||||
if len(files) > MAX_UPLOAD_FILES:
|
||||
raise HTTPException(400, f"Máximo {MAX_UPLOAD_FILES} arquivos por mensagem")
|
||||
|
||||
# Process uploaded files into attachments
|
||||
attachments = []
|
||||
file_names = []
|
||||
for f in files:
|
||||
@@ -2429,7 +2444,6 @@ async def chat_with_files(
|
||||
elif mime in DOC_MIMES:
|
||||
attachments.append({"type": "document", "mime": mime, "data_uri": data_uri})
|
||||
else:
|
||||
# Text-based files: read content and append to message
|
||||
try:
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
@@ -2437,7 +2451,6 @@ async def chat_with_files(
|
||||
message = f"{message}\n\n--- Conteúdo de {f.filename} ---\n{text[:50000]}"
|
||||
file_names.append(f.filename)
|
||||
|
||||
# Build a synthetic ChatMsg and delegate to the chat logic
|
||||
msg = ChatMsg(
|
||||
message=message or "Analise os arquivos anexados.",
|
||||
session_id=session_id,
|
||||
@@ -2455,19 +2468,35 @@ async def chat_with_files(
|
||||
presence_penalty=presence_penalty,
|
||||
)
|
||||
|
||||
# Store attachments in request state for _call_genai
|
||||
result = await _chat_core_impl(msg, u, attachments=attachments if attachments else None)
|
||||
sid, mid_or_result, genai_cfg = _chat_start(msg, u, attachments=attachments if attachments else None)
|
||||
if genai_cfg is None:
|
||||
return mid_or_result # immediate fallback response
|
||||
|
||||
if file_names:
|
||||
# Save file reference in user message
|
||||
with db() as c:
|
||||
c.execute("UPDATE chat_messages SET content = content || ? WHERE session_id=? AND role='user' ORDER BY created_at DESC LIMIT 1",
|
||||
(f"\n[📎 {', '.join(file_names)}]", result["session_id"]))
|
||||
return result
|
||||
c.execute("UPDATE chat_messages SET content = content || ? WHERE session_id=? AND role='user' AND status='done' ORDER BY created_at DESC LIMIT 1",
|
||||
(f"\n[📎 {', '.join(file_names)}]", sid))
|
||||
|
||||
bg.add_task(_chat_background, mid_or_result, sid, msg, dict(u), genai_cfg, attachments if attachments else None)
|
||||
return {"message_id": mid_or_result, "session_id": sid, "status": "processing"}
|
||||
|
||||
|
||||
@app.post("/api/chat")
|
||||
async def chat(msg: ChatMsg, u=Depends(current_user)):
|
||||
return await _chat_core_impl(msg, u)
|
||||
async def chat(msg: ChatMsg, bg: BackgroundTasks, u=Depends(current_user)):
|
||||
sid, mid_or_result, genai_cfg = _chat_start(msg, u)
|
||||
if genai_cfg is None:
|
||||
return mid_or_result # immediate fallback response
|
||||
bg.add_task(_chat_background, mid_or_result, sid, msg, dict(u), genai_cfg)
|
||||
return {"message_id": mid_or_result, "session_id": sid, "status": "processing"}
|
||||
|
||||
|
||||
@app.get("/api/chat/{mid}/status")
|
||||
async def chat_message_status(mid: str, u=Depends(current_user)):
|
||||
with db() as c:
|
||||
r = c.execute("SELECT id, session_id, role, content, model_id, status FROM chat_messages WHERE id=?", (mid,)).fetchone()
|
||||
if not r:
|
||||
raise HTTPException(404, "Message not found")
|
||||
return dict(r)
|
||||
|
||||
|
||||
def _agent_respond(msg, user):
|
||||
@@ -2499,27 +2528,6 @@ async def clear_chat(sid, u=Depends(current_user)):
|
||||
with db() as c: c.execute("DELETE FROM chat_messages WHERE session_id=? AND user_id=?", (sid, u["id"]))
|
||||
return {"ok": True}
|
||||
|
||||
# ── Downloads ─────────────────────────────────────────────────────────────────
|
||||
@app.get("/api/downloads")
|
||||
async def list_dl(u=Depends(current_user)):
|
||||
with db() as c:
|
||||
q="SELECT id,tenancy_name,status,created_at,completed_at FROM reports WHERE status='completed'"
|
||||
rows=c.execute(q+" ORDER BY created_at DESC").fetchall() if u["role"]=="admin" \
|
||||
else c.execute(q+" AND user_id=? ORDER BY created_at DESC",(u["id"],)).fetchall()
|
||||
res=[]
|
||||
for r in rows:
|
||||
d=dict(r); d["files"]=[]
|
||||
rd=REPORTS/d["id"]
|
||||
if rd.exists(): d["files"]=[{"name":f.name,"size":f.stat().st_size} for f in rd.iterdir() if f.is_file()]
|
||||
res.append(d)
|
||||
return res
|
||||
|
||||
@app.get("/api/downloads/{rid}/{fname}")
|
||||
async def dl_file(rid, fname, u=Depends(current_user)):
|
||||
p=REPORTS/rid/fname
|
||||
if not p.exists(): raise HTTPException(404)
|
||||
return FileResponse(p, filename=fname)
|
||||
|
||||
# ── Audit ─────────────────────────────────────────────────────────────────────
|
||||
@app.get("/api/audit-log")
|
||||
async def audit_log(limit:int=Query(100,le=500), u=Depends(require("admin"))):
|
||||
@@ -2544,19 +2552,6 @@ async def get_config_logs(
|
||||
with db() as c: rows = c.execute(query, params).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
@app.delete("/api/config-logs")
|
||||
async def clear_config_logs(
|
||||
config_type: str = Query(None), config_id: str = Query(None),
|
||||
u=Depends(require("admin"))
|
||||
):
|
||||
query = "DELETE FROM config_logs WHERE 1=1"
|
||||
params = []
|
||||
if config_type: query += " AND config_type=?"; params.append(config_type)
|
||||
if config_id: query += " AND config_id=?"; params.append(config_id)
|
||||
with db() as c: c.execute(query, params)
|
||||
return {"ok": True}
|
||||
|
||||
# ── Health ────────────────────────────────────────────────────────────────────
|
||||
# ── App Settings ──────────────────────────────────────────────────────────────
|
||||
@app.get("/api/settings/{key}")
|
||||
async def get_setting(key: str, u=Depends(current_user)):
|
||||
|
||||
Reference in New Issue
Block a user