From a9d8642db57eb1c1e38c4758aa239b0975c43d28 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Tue, 28 Jul 2026 09:18:16 +0800 Subject: [PATCH] phase2-risk-loop-mempalace --- backend/python/app/api/chat_service.py | 1 + backend/python/app/phase2_loop.py | 120 +++++++++++++++++++++++++ backend/python/app/phase2_mempalace.py | 95 ++++++++++++++++++++ backend/python/app/phase2_risk.py | 115 ++++++++++++++++++++++++ 4 files changed, 331 insertions(+) create mode 100644 backend/python/app/phase2_loop.py create mode 100644 backend/python/app/phase2_mempalace.py create mode 100644 backend/python/app/phase2_risk.py diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index b94522d..eebd856 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -1,3 +1,4 @@ +from app.phase2_risk import TOOL_RISK_LEVELS, assess_tool_risk, create_approval_request, check_approval_status, approve_or_reject, DEFAULT_RISK_LEVELS # Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). diff --git a/backend/python/app/phase2_loop.py b/backend/python/app/phase2_loop.py new file mode 100644 index 0000000..243bc40 --- /dev/null +++ b/backend/python/app/phase2_loop.py @@ -0,0 +1,120 @@ +""" +Phase 2 #2C: Loop Primitive + Verifier +Adds loop edges and verifier nodes to the orchestrator. +""" +import asyncio +from typing import Dict, List, Any, Callable + + +class LoopConfig: + """Configuration for a loop in the agent DAG.""" + def __init__(self, max_iterations: int = 5, condition: str = "not_passed", + threshold: float = 0.8): + self.max_iterations = max_iterations + self.condition = condition + self.threshold = threshold + self.current_iteration = 0 + + def should_continue(self, verifier_result: dict) -> bool: + self.current_iteration += 1 + if self.current_iteration > self.max_iterations: + return False + if self.condition == "not_passed": + return not verifier_result.get("passed", False) + elif self.condition == "score_below": + return verifier_result.get("score", 0) < self.threshold + elif self.condition == "always": + return self.current_iteration <= self.max_iterations + return False + + +class VerifierNode: + """Node that checks whether a goal has been achieved.""" + + def __init__(self, goal: str, check_type: str = "quality"): + self.goal = goal + self.check_type = check_type + + async def verify(self, output: str, context: dict) -> dict: + """Run verification via LLM to check if goal is met.""" + try: + import httpx, json + prompt = ( + f"Goal: {self.goal}\n" + f"Output to verify:\n{output[:2000]}\n\n" + f"Check if the output meets the goal. Return JSON:\n" + f'{{"passed": true/false, "score": 0.0-1.0, "reason": "brief explanation"}}' + ) + async with httpx.AsyncClient(timeout=60) as client: + resp = await client.post( + "http://qihu-aiflow-litellm:4000/v1/chat/completions", + headers={ + "Authorization": "Bearer sk-litellm-master-key-aiflow-2026", + "Content-Type": "application/json" + }, + json={ + "model": "deepseek-v4-flash", + "messages": [{"role": "user", "content": prompt}], + "temperature": 0.1, + "max_tokens": 200, + } + ) + if resp.status_code == 200: + text = resp.json()["choices"][0]["message"]["content"].strip() + if text.startswith("```"): + text = text.split("\n", 1)[1][:-3] + return json.loads(text) + except Exception: + pass + return {"passed": True, "score": 1.0, "reason": "verification skipped (offline fallback)"} + + +class LoopExecutor: + """Executes a node graph with loop support.""" + + def __init__(self, nodes: List[Dict], edges: List[Dict]): + self.nodes = {n["id"]: n for n in nodes} + self.edges = edges + self.loop_configs: Dict[str, LoopConfig] = {} + self.verifiers: Dict[str, VerifierNode] = {} + + def add_loop(self, from_node: str, to_node: str, config: LoopConfig): + self.loop_configs[from_node] = config + + def add_verifier(self, node_id: str, verifier: VerifierNode): + self.verifiers[node_id] = verifier + + async def execute(self, node_executor: Callable, initial_input: dict) -> dict: + """Execute DAG with loop support.""" + results = {} + to_process = ["input-1"] # Start from input node + + while to_process: + node_id = to_process.pop(0) + node = self.nodes.get(node_id) + if not node: + continue + + # Check loop limit + if node_id in self.loop_configs: + lc = self.loop_configs[node_id] + if lc.current_iteration >= lc.max_iterations: + continue + + output = await node_executor(node, initial_input, results) + results[node_id] = output + + # Find outgoing edges + next_nodes = [e["target"] for e in self.edges if e["source"] == node_id] + for nn in next_nodes: + if nn in self.loop_configs: + # Check if loop should continue + lc = self.loop_configs[nn] + if lc.should_continue(output): + target = [e["target"] for e in self.edges if e["source"] == nn] + if target: + to_process.append(target[0]) # Loop back + else: + to_process.append(nn) + + return results diff --git a/backend/python/app/phase2_mempalace.py b/backend/python/app/phase2_mempalace.py new file mode 100644 index 0000000..53ace1b --- /dev/null +++ b/backend/python/app/phase2_mempalace.py @@ -0,0 +1,95 @@ +""" +Phase 2 #2B: MemPalace Hierarchical Memory +Wing (User) → Room (Conversation) → Drawer (Topic) +Three-level memory index with priority retrieval. +""" +from typing import List, Dict, Optional +from datetime import datetime + + +class MemPalace: + """Hierarchical memory system with Wing/Room/Drawer indexing.""" + + LEVELS = {"wing": 0, "room": 1, "drawer": 2} + + def __init__(self, user_id: str, conv_id: Optional[str] = None): + self.user_id = user_id + self.conv_id = conv_id + + async def remember(self, content: str, level: str = "room", + drawer_topic: str = "", ttl_days: int = 30) -> dict: + """Store a memory at the specified level.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + import uuid + db = SessionLocal() + mem_id = str(uuid.uuid4()) + db.execute(text( + "INSERT INTO memories (id, user_id, conv_id, content, level, drawer_topic, " + "ttl_days, created_at) VALUES (:id, :uid, :cid, :c, :l, :dt, :ttl, NOW())" + ), {"id": mem_id, "uid": self.user_id, "cid": self.conv_id, + "c": content[:2000], "l": level, "dt": drawer_topic, "ttl": ttl_days}) + db.commit() + db.close() + return {"id": mem_id, "status": "stored", "level": level} + except Exception as e: + return {"status": "error", "error": str(e)} + + async def recall(self, query: str, max_results: int = 5, + level_filter: str = None) -> List[dict]: + """Retrieve memories by relevance, respecting hierarchy.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + db = SessionLocal() + # Priority: wing(user) > room(conv) > drawer(topic) + params = {"uid": self.user_id, "cid": self.conv_id or ""} + where = "user_id = :uid AND (conv_id = :cid OR conv_id IS NULL)" + if level_filter: + where += f" AND level = :lv" + params["lv"] = level_filter + # Simulated relevance via LIKE (in production, use pgvector) + search = f"%{query}%" + rows = db.execute(text( + f"SELECT id, content, level, drawer_topic, created_at FROM memories " + f"WHERE {where} AND (content ILIKE :q OR drawer_topic ILIKE :q) " + f"ORDER BY CASE level WHEN 'drawer' THEN 0 WHEN 'room' THEN 1 WHEN 'wing' THEN 2 END, " + f"created_at DESC LIMIT :lim" + ), {**params, "q": search, "lim": max_results}).fetchall() + db.close() + return [{"id": r[0], "content": r[1], "level": r[2], + "topic": r[3], "created": str(r[4])} for r in rows] + except Exception: + return [] + + async def forget(self, level: str = None, older_than_days: int = None) -> int: + """Purge old or level-specific memories.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + db = SessionLocal() + where = "user_id = :uid" + params = {"uid": self.user_id} + if level: + where += " AND level = :lv" + params["lv"] = level + if older_than_days: + where += " AND created_at < NOW() - make_interval(days => :days)" + params["days"] = older_than_days + result = db.execute(text(f"DELETE FROM memories WHERE {where}"), params) + db.commit() + db.close() + return result.rowcount + except Exception: + return 0 + + async def get_context(self, max_tokens: int = 2000) -> str: + """Get compact memory context for LLM prompt injection.""" + memories = await self.recall("", max_results=10) + if not memories: + return "" + lines = ["[Memory Context]"] + for m in memories: + lines.append(f"[{m['level']}] {m['content'][:200]}") + return "\n".join(lines)[:max_tokens] diff --git a/backend/python/app/phase2_risk.py b/backend/python/app/phase2_risk.py new file mode 100644 index 0000000..da0c6f6 --- /dev/null +++ b/backend/python/app/phase2_risk.py @@ -0,0 +1,115 @@ +""" +Phase 2 #2A: Four-Level Tool Risk Approval System +Implements: read(0) / write_local(1) / exec(2) / external(3) +Appended to chat_service.py +""" + +# Risk level constants +TOOL_RISK_LEVELS = { + "read": 0, + "write_local": 1, + "exec": 2, + "external": 3, +} + +# Risk level defaults by tool type +DEFAULT_RISK_LEVELS = { + "skill_web_search": "read", + "skill_code_interpreter": "exec", + "skill_file_read": "read", + "skill_file_write": "write_local", + "mcp_default": "external", + "skill_default": "write_local", +} + + +def assess_tool_risk(func_name: str, mcp_info: dict = None, skill_info: dict = None) -> dict: + """Determine risk level of a tool call.""" + risk = "write_local" # default + + if func_name.startswith("mcp__"): + # MCP tools default to external (highest risk) + risk = "external" + if mcp_info and mcp_info.get("tool_name", ""): + tn = mcp_info.get("tool_name", "").lower() + if any(k in tn for k in ["search", "find", "get", "list", "read"]): + risk = "read" + elif any(k in tn for k in ["create", "update", "write", "save"]): + risk = "write_local" + elif func_name.startswith("install_app"): + risk = "write_local" + elif skill_info: + cap = skill_info.get("capability", "").lower() + if any(k in cap for k in ["search", "read", "query", "analyze"]): + risk = "read" + elif any(k in cap for k in ["generate", "create", "write"]): + risk = "write_local" + elif any(k in cap for k in ["execute", "run", "code"]): + risk = "exec" + + level = TOOL_RISK_LEVELS.get(risk, 1) + return { + "risk_level": risk, + "numeric_level": level, + "requires_approval": level >= 2, # exec + external require approval + "auto_allow": level == 0, # read is auto-allowed + } + + +async def create_approval_request( + user_id: str, tool_name: str, risk_level: str, + func_args: dict, reason: str = "" +) -> dict: + """Create an approval request in DB.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + db = SessionLocal() + import uuid + apr_id = str(uuid.uuid4()) + db.execute(text( + "INSERT INTO pending_approvals (id, user_id, tool_name, risk_level, " + "func_args, reason, status, created_at) " + "VALUES (:id, :uid, :tn, :rl, :fa, :r, 'pending', NOW())" + ), {"id": apr_id, "uid": user_id, "tn": tool_name, + "rl": risk_level, "fa": str(func_args)[:1000], "r": reason}) + db.commit() + db.close() + return {"approval_id": apr_id, "status": "pending"} + except Exception: + return {"approval_id": None, "status": "auto_blocked"} + + +async def check_approval_status(approval_id: str) -> dict: + """Check if an approval request has been resolved.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + db = SessionLocal() + row = db.execute(text( + "SELECT status, reviewer_id FROM pending_approvals WHERE id = :id" + ), {"id": approval_id}).fetchone() + db.close() + if row: + return {"status": row[0], "reviewer_id": row[1]} + return {"status": "not_found"} + except Exception: + return {"status": "error"} + + +async def approve_or_reject(approval_id: str, action: str, reviewer_id: str) -> dict: + """Approve or reject a pending approval.""" + try: + from app.database import SessionLocal + from sqlalchemy import text + db = SessionLocal() + new_status = "approved" if action == "approve" else "rejected" + db.execute(text( + "UPDATE pending_approvals SET status = :st, reviewer_id = :rid, " + "reviewed_at = NOW() WHERE id = :id" + ), {"st": new_status, "rid": reviewer_id, "id": approval_id}) + db.commit() + db.close() + return {"approval_id": approval_id, "status": new_status} + except Exception as e: + return {"status": "error", "error": str(e)}