""" KnowHub Server Agent 工具使用经验的共享平台。 FastAPI + SQLite,单文件部署。 """ import os import re import json import sqlite3 import asyncio import base64 from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Optional from pathlib import Path from cryptography.hazmat.primitives.ciphers.aead import AESGCM from fastapi import FastAPI, HTTPException, Query, Header from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field # 导入 LLM 调用(需要 agent 模块在 Python path 中) import sys sys.path.insert(0, str(Path(__file__).parent.parent)) # 加载环境变量 from dotenv import load_dotenv load_dotenv(Path(__file__).parent.parent / ".env") from agent.llm.openrouter import openrouter_llm_call BRAND_NAME = os.getenv("BRAND_NAME", "KnowHub") BRAND_API_ENV = os.getenv("BRAND_API_ENV", "KNOWHUB_API") BRAND_DB = os.getenv("BRAND_DB", "knowhub.db") # 组织密钥配置(格式:org1:key1_base64,org2:key2_base64) ORG_KEYS_RAW = os.getenv("ORG_KEYS", "") ORG_KEYS = {} if ORG_KEYS_RAW: for pair in ORG_KEYS_RAW.split(","): if ":" in pair: org, key_b64 = pair.split(":", 1) ORG_KEYS[org.strip()] = key_b64.strip() DB_PATH = Path(__file__).parent / BRAND_DB # --- 数据库 --- def get_db() -> sqlite3.Connection: conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn # --- 加密/解密 --- def get_org_key(resource_id: str) -> Optional[bytes]: """从content_id提取组织前缀,返回对应密钥""" if "/" in resource_id: org = resource_id.split("/")[0] if org in ORG_KEYS: return base64.b64decode(ORG_KEYS[org]) return None def encrypt_content(resource_id: str, plaintext: str) -> str: """加密内容,返回格式:encrypted:AES256-GCM:{base64_data}""" if not plaintext: return "" key = get_org_key(resource_id) if not key: # 没有配置密钥,明文存储(不推荐) return plaintext aesgcm = AESGCM(key) nonce = os.urandom(12) # 96-bit nonce ciphertext = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None) # 组合 nonce + ciphertext encrypted_data = nonce + ciphertext encoded = base64.b64encode(encrypted_data).decode("ascii") return f"encrypted:AES256-GCM:{encoded}" def decrypt_content(resource_id: str, encrypted_text: str, provided_key: Optional[str] = None) -> str: """解密内容,如果没有提供密钥或密钥错误,返回[ENCRYPTED]""" if not encrypted_text: return "" if not encrypted_text.startswith("encrypted:AES256-GCM:"): # 未加密的内容,直接返回 return encrypted_text # 提取加密数据 encoded = encrypted_text.split(":", 2)[2] encrypted_data = base64.b64decode(encoded) nonce = encrypted_data[:12] ciphertext = encrypted_data[12:] # 获取密钥 key = None if provided_key: # 使用提供的密钥 try: key = base64.b64decode(provided_key) except Exception: return "[ENCRYPTED]" else: # 从配置中获取 key = get_org_key(resource_id) if not key: return "[ENCRYPTED]" try: aesgcm = AESGCM(key) plaintext = aesgcm.decrypt(nonce, ciphertext, None) return plaintext.decode("utf-8") except Exception: return "[ENCRYPTED]" def init_db(): conn = get_db() conn.execute(""" CREATE TABLE IF NOT EXISTS experiences ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, url TEXT DEFAULT '', category TEXT DEFAULT '', task TEXT NOT NULL, score INTEGER CHECK(score BETWEEN 1 AND 5), outcome TEXT DEFAULT '', tips TEXT DEFAULT '', content_id TEXT DEFAULT '', submitted_by TEXT DEFAULT '', created_at TEXT NOT NULL ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_name ON experiences(name)") conn.execute(""" CREATE TABLE IF NOT EXISTS resources ( id TEXT PRIMARY KEY, title TEXT DEFAULT '', body TEXT NOT NULL, secure_body TEXT DEFAULT '', content_type TEXT DEFAULT 'text', metadata TEXT DEFAULT '{}', sort_order INTEGER DEFAULT 0, submitted_by TEXT DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT DEFAULT '' ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS knowledge ( id TEXT PRIMARY KEY, message_id TEXT DEFAULT '', types TEXT NOT NULL, -- JSON array: ["strategy", "tool"] task TEXT NOT NULL, tags TEXT DEFAULT '{}', -- JSON object: {"category": "...", "domain": "..."} scopes TEXT DEFAULT '["org:cybertogether"]', -- JSON array owner TEXT DEFAULT '', content TEXT NOT NULL, resource_ids TEXT DEFAULT '[]', -- JSON array: ["code/selenium/login", "credentials/website"] source TEXT DEFAULT '{}', -- JSON object: {name, category, urls, agent_id, submitted_by, timestamp} eval TEXT DEFAULT '{}', -- JSON object: {score, helpful, harmful, confidence, histories} created_at TEXT NOT NULL, updated_at TEXT DEFAULT '' ) """) conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_types ON knowledge(types)") conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_task ON knowledge(task)") conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_owner ON knowledge(owner)") conn.execute("CREATE INDEX IF NOT EXISTS idx_knowledge_scopes ON knowledge(scopes)") conn.commit() conn.close() # --- Models --- class ResourceIn(BaseModel): id: str title: str = "" body: str secure_body: str = "" content_type: str = "text" # text|code|credential|cookie metadata: dict = {} sort_order: int = 0 submitted_by: str = "" class ResourcePatchIn(BaseModel): """PATCH /api/resource/{id} 请求体""" title: Optional[str] = None body: Optional[str] = None secure_body: Optional[str] = None content_type: Optional[str] = None metadata: Optional[dict] = None # Knowledge Models class KnowledgeIn(BaseModel): task: str content: str types: list[str] = ["strategy"] tags: dict = {} scopes: list[str] = ["org:cybertogether"] owner: str = "" message_id: str = "" resource_ids: list[str] = [] source: dict = {} # {name, category, urls, agent_id, submitted_by, timestamp} eval: dict = {} # {score, helpful, harmful, confidence} class KnowledgeOut(BaseModel): id: str message_id: str types: list[str] task: str tags: dict scopes: list[str] owner: str content: str resource_ids: list[str] source: dict eval: dict created_at: str updated_at: str class KnowledgeUpdateIn(BaseModel): add_helpful_case: Optional[dict] = None add_harmful_case: Optional[dict] = None update_score: Optional[int] = Field(default=None, ge=1, le=5) evolve_feedback: Optional[str] = None class KnowledgePatchIn(BaseModel): """PATCH /api/knowledge/{id} 请求体(直接字段编辑)""" task: Optional[str] = None content: Optional[str] = None types: Optional[list[str]] = None tags: Optional[dict] = None scopes: Optional[list[str]] = None owner: Optional[str] = None class MessageExtractIn(BaseModel): """POST /api/extract 请求体(消息历史提取)""" messages: list[dict] # [{role: str, content: str}, ...] agent_id: str = "unknown" submitted_by: str # 必填,作为 owner session_key: str = "" class KnowledgeBatchUpdateIn(BaseModel): feedback_list: list[dict] class KnowledgeSearchResponse(BaseModel): results: list[dict] count: int class ResourceNode(BaseModel): id: str title: str class ResourceOut(BaseModel): id: str title: str body: str secure_body: str = "" content_type: str = "text" metadata: dict = {} toc: Optional[ResourceNode] = None children: list[ResourceNode] prev: Optional[ResourceNode] = None next: Optional[ResourceNode] = None # --- App --- @asynccontextmanager async def lifespan(app: FastAPI): init_db() yield app = FastAPI(title=BRAND_NAME, lifespan=lifespan) # --- Knowledge API --- @app.post("/api/resource", status_code=201) def submit_resource(resource: ResourceIn): conn = get_db() try: now = datetime.now(timezone.utc).isoformat() # 加密敏感内容 encrypted_secure_body = encrypt_content(resource.id, resource.secure_body) conn.execute( "INSERT OR REPLACE INTO resources" "(id, title, body, secure_body, content_type, metadata, sort_order, submitted_by, created_at, updated_at)" " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( resource.id, resource.title, resource.body, encrypted_secure_body, resource.content_type, json.dumps(resource.metadata), resource.sort_order, resource.submitted_by, now, now, ), ) conn.commit() return {"status": "ok", "id": resource.id} finally: conn.close() @app.get("/api/resource/{resource_id:path}", response_model=ResourceOut) def get_resource(resource_id: str, x_org_key: Optional[str] = Header(None)): conn = get_db() try: row = conn.execute( "SELECT id, title, body, secure_body, content_type, metadata, sort_order FROM resources WHERE id = ?", (resource_id,), ).fetchone() if not row: raise HTTPException(status_code=404, detail=f"Resource not found: {resource_id}") # 解密敏感内容 secure_body = decrypt_content(resource_id, row["secure_body"] or "", x_org_key) # 解析metadata metadata = json.loads(row["metadata"] or "{}") # 计算导航上下文 root_id = resource_id.split("/")[0] if "/" in resource_id else resource_id # TOC (根节点) toc = None if "/" in resource_id: toc_row = conn.execute( "SELECT id, title FROM resources WHERE id = ?", (root_id,), ).fetchone() if toc_row: toc = ResourceNode(id=toc_row["id"], title=toc_row["title"]) # Children (子节点) children = [] children_rows = conn.execute( "SELECT id, title FROM resources WHERE id LIKE ? AND id != ? ORDER BY sort_order", (f"{resource_id}/%", resource_id), ).fetchall() children = [ResourceNode(id=r["id"], title=r["title"]) for r in children_rows] # Prev/Next (同级节点) prev_node = None next_node = None if "/" in resource_id: siblings = conn.execute( "SELECT id, title, sort_order FROM resources WHERE id LIKE ? AND id NOT LIKE ? ORDER BY sort_order", (f"{root_id}/%", f"{root_id}/%/%"), ).fetchall() for i, sib in enumerate(siblings): if sib["id"] == resource_id: if i > 0: prev_node = ResourceNode(id=siblings[i-1]["id"], title=siblings[i-1]["title"]) if i < len(siblings) - 1: next_node = ResourceNode(id=siblings[i+1]["id"], title=siblings[i+1]["title"]) break return ResourceOut( id=row["id"], title=row["title"], body=row["body"], secure_body=secure_body, content_type=row["content_type"], metadata=metadata, toc=toc, children=children, prev=prev_node, next=next_node, ) finally: conn.close() @app.patch("/api/resource/{resource_id:path}") def patch_resource(resource_id: str, patch: ResourcePatchIn): """更新resource字段""" conn = get_db() try: # 检查是否存在 row = conn.execute("SELECT id FROM resources WHERE id = ?", (resource_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail=f"Resource not found: {resource_id}") # 构建更新语句 updates = [] params = [] if patch.title is not None: updates.append("title = ?") params.append(patch.title) if patch.body is not None: updates.append("body = ?") params.append(patch.body) if patch.secure_body is not None: encrypted = encrypt_content(resource_id, patch.secure_body) updates.append("secure_body = ?") params.append(encrypted) if patch.content_type is not None: updates.append("content_type = ?") params.append(patch.content_type) if patch.metadata is not None: updates.append("metadata = ?") params.append(json.dumps(patch.metadata)) if not updates: return {"status": "ok", "message": "No fields to update"} # 添加updated_at updates.append("updated_at = ?") params.append(datetime.now(timezone.utc).isoformat()) # 执行更新 params.append(resource_id) sql = f"UPDATE resources SET {', '.join(updates)} WHERE id = ?" conn.execute(sql, params) conn.commit() return {"status": "ok", "id": resource_id} finally: conn.close() @app.get("/api/resource") def list_resources( content_type: Optional[str] = Query(None), limit: int = Query(100, ge=1, le=1000) ): """列出所有resource""" conn = get_db() try: sql = "SELECT id, title, content_type, metadata, created_at FROM resources" params = [] if content_type: sql += " WHERE content_type = ?" params.append(content_type) sql += " ORDER BY id LIMIT ?" params.append(limit) rows = conn.execute(sql, params).fetchall() results = [] for row in rows: results.append({ "id": row["id"], "title": row["title"], "content_type": row["content_type"], "metadata": json.loads(row["metadata"] or "{}"), "created_at": row["created_at"], }) return {"results": results, "count": len(results)} finally: conn.close() # --- Knowledge API --- # ===== Knowledge API ===== # 两阶段检索逻辑 async def _route_knowledge_by_llm(query_text: str, metadata_list: list[dict], k: int = 5) -> list[str]: """ 第一阶段:语义路由。 让 LLM 挑选出 2*k 个语义相关的 ID。 """ if not metadata_list: return [] routing_k = k * 2 routing_data = [ { "id": m["id"], "types": m["types"], "task": m["task"][:100] } for m in metadata_list ] prompt = f""" 你是一个知识检索专家。根据用户的当前任务需求,从下列原子知识元数据中挑选出最相关的最多 {routing_k} 个知识 ID。 任务需求:"{query_text}" 可选知识列表: {json.dumps(routing_data, ensure_ascii=False, indent=1)} 请直接输出 ID 列表,用逗号分隔(例如: knowledge-20260302-001, research-20260302-002)。若无相关项请输出 "None"。 """ try: print(f"\n[Step 1: 知识语义路由] 任务: '{query_text}' | 候选总数: {len(metadata_list)} | 目标提取数: {routing_k}") response = await openrouter_llm_call( messages=[{"role": "user", "content": prompt}], model="google/gemini-2.5-flash-lite" ) content = response.get("content", "").strip() selected_ids = [idx.strip() for idx in re.split(r'[,\s]+', content) if idx.strip().startswith(("knowledge-", "research-"))] print(f"[Step 1: 知识语义路由] LLM 初选 ID ({len(selected_ids)}个): {selected_ids}") return selected_ids except Exception as e: print(f"LLM 知识路由失败: {e}") return [] async def _search_knowledge_two_stage( query_text: str, top_k: int = 5, min_score: int = 3, types_filter: Optional[list[str]] = None, owner_filter: Optional[str] = None, conn: sqlite3.Connection = None ) -> list[dict]: """ 两阶段检索:语义路由 + 质量精排 """ if conn is None: conn = get_db() should_close = True else: should_close = False try: # 阶段 1: 解析所有知识 query = "SELECT * FROM knowledge" rows = conn.execute(query).fetchall() if not rows: return [] content_map = {} metadata_list = [] for row in rows: kid = row["id"] types = json.loads(row["types"]) # 标签过滤 if types_filter: if not any(t in types for t in types_filter): continue # owner 过滤 if owner_filter and row["owner"] != owner_filter: continue task = row["task"] content_text = row["content"] eval_data = json.loads(row["eval"]) source = json.loads(row["source"]) meta_item = { "id": kid, "types": types, "task": task, "score": eval_data.get("score", 3), "helpful": eval_data.get("helpful", 0), "harmful": eval_data.get("harmful", 0), } metadata_list.append(meta_item) content_map[kid] = { "task": task, "content": content_text, "types": types, "tags": json.loads(row["tags"]), "scopes": json.loads(row["scopes"]), "owner": row["owner"], "score": meta_item["score"], "helpful": meta_item["helpful"], "harmful": meta_item["harmful"], "message_id": row["message_id"], "source": source, "eval": eval_data, "created_at": row["created_at"], "updated_at": row["updated_at"] } if not metadata_list: return [] # 阶段 2: 语义路由 (取 2*k) candidate_ids = await _route_knowledge_by_llm(query_text, metadata_list, k=top_k) # 阶段 3: 质量精排 print(f"[Step 2: 知识质量精排] 正在根据评分和反馈进行打分...") scored_items = [] for kid in candidate_ids: if kid in content_map: item = content_map[kid] score = item["score"] helpful = item["helpful"] harmful = item["harmful"] # 计算综合分:基础分 + helpful - harmful*2 quality_score = score + helpful - (harmful * 2.0) # 过滤门槛 if score < min_score or quality_score < 0: print(f" - 剔除低质量知识: {kid} (Score: {score}, Helpful: {helpful}, Harmful: {harmful})") continue scored_items.append({ "id": kid, "message_id": item["message_id"], "types": item["types"], "task": item["task"], "tags": item["tags"], "scopes": item["scopes"], "owner": item["owner"], "content": item["content"], "source": item["source"], "eval": item["eval"], "quality_score": quality_score, "created_at": item["created_at"], "updated_at": item["updated_at"] }) # 按照质量分排序 final_sorted = sorted(scored_items, key=lambda x: x["quality_score"], reverse=True) # 截取最终的 top_k result = final_sorted[:top_k] print(f"[Step 2: 知识质量精排] 最终选定知识: {[it['id'] for it in result]}") print(f"[Knowledge System] 检索结束。\n") return result finally: if should_close: conn.close() @app.get("/api/knowledge/search") async def search_knowledge_api( q: str = Query(..., description="查询文本"), top_k: int = Query(default=5, ge=1, le=20), min_score: int = Query(default=3, ge=1, le=5), types: Optional[str] = None, owner: Optional[str] = None ): """检索知识(两阶段:语义路由 + 质量精排)""" conn = get_db() try: types_filter = types.split(",") if types else None results = await _search_knowledge_two_stage( query_text=q, top_k=top_k, min_score=min_score, types_filter=types_filter, owner_filter=owner, conn=conn ) return {"results": results, "count": len(results)} finally: conn.close() @app.post("/api/knowledge", status_code=201) def save_knowledge(knowledge: KnowledgeIn): """保存新知识""" import uuid conn = get_db() try: # 生成 ID timestamp = datetime.now().strftime('%Y%m%d-%H%M%S') random_suffix = uuid.uuid4().hex[:4] knowledge_id = f"knowledge-{timestamp}-{random_suffix}" now = datetime.now(timezone.utc).isoformat() # 设置默认值 owner = knowledge.owner or f"agent:{knowledge.source.get('agent_id', 'unknown')}" # 准备 source source = { "name": knowledge.source.get("name", ""), "category": knowledge.source.get("category", ""), "urls": knowledge.source.get("urls", []), "agent_id": knowledge.source.get("agent_id", "unknown"), "submitted_by": knowledge.source.get("submitted_by", ""), "timestamp": now, "message_id": knowledge.message_id } # 准备 eval eval_data = { "score": knowledge.eval.get("score", 3), "helpful": knowledge.eval.get("helpful", 1), "harmful": knowledge.eval.get("harmful", 0), "confidence": knowledge.eval.get("confidence", 0.5), "helpful_history": [], "harmful_history": [] } conn.execute( """INSERT INTO knowledge (id, message_id, types, task, tags, scopes, owner, content, resource_ids, source, eval, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( knowledge_id, knowledge.message_id, json.dumps(knowledge.types), knowledge.task, json.dumps(knowledge.tags), json.dumps(knowledge.scopes), owner, knowledge.content, json.dumps(knowledge.resource_ids), json.dumps(source), json.dumps(eval_data), now, now, ), ) conn.commit() return {"status": "ok", "knowledge_id": knowledge_id} finally: conn.close() @app.get("/api/knowledge") def list_knowledge( limit: int = Query(default=100, ge=1, le=1000), types: Optional[str] = None, scopes: Optional[str] = None, owner: Optional[str] = None, tags: Optional[str] = None ): """列出知识(支持后端筛选)""" conn = get_db() try: query = "SELECT * FROM knowledge" params = [] conditions = [] # types 支持多个,用 AND 连接(交集:必须同时包含所有选中的type) if types: type_list = [t.strip() for t in types.split(',') if t.strip()] if type_list: for t in type_list: conditions.append("types LIKE ?") params.append(f"%{t}%") if scopes: conditions.append("scopes LIKE ?") params.append(f"%{scopes}%") if owner: conditions.append("owner LIKE ?") params.append(f"%{owner}%") # tags 支持多个,用 AND 连接(交集:必须同时包含所有选中的tag) if tags: tag_list = [t.strip() for t in tags.split(',') if t.strip()] if tag_list: for t in tag_list: conditions.append("tags LIKE ?") params.append(f"%{t}%") if conditions: query += " WHERE " + " AND ".join(conditions) query += " ORDER BY created_at DESC LIMIT ?" params.append(limit) rows = conn.execute(query, params).fetchall() results = [] for row in rows: results.append({ "id": row["id"], "message_id": row["message_id"], "types": json.loads(row["types"]), "task": row["task"], "tags": json.loads(row["tags"]), "scopes": json.loads(row["scopes"]), "owner": row["owner"], "content": row["content"], "source": json.loads(row["source"]), "eval": json.loads(row["eval"]), "created_at": row["created_at"], "updated_at": row["updated_at"] }) return {"results": results, "count": len(results)} finally: conn.close() @app.get("/api/knowledge/meta/tags") def get_all_tags(): """获取所有已有的 tags""" conn = get_db() try: rows = conn.execute("SELECT tags FROM knowledge").fetchall() all_tags = set() for row in rows: tags_dict = json.loads(row["tags"]) for key in tags_dict.keys(): all_tags.add(key) return {"tags": sorted(list(all_tags))} finally: conn.close() @app.get("/api/knowledge/{knowledge_id}") def get_knowledge(knowledge_id: str): """获取单条知识""" conn = get_db() try: row = conn.execute( "SELECT * FROM knowledge WHERE id = ?", (knowledge_id,) ).fetchone() if not row: raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}") return { "id": row["id"], "message_id": row["message_id"], "types": json.loads(row["types"]), "task": row["task"], "tags": json.loads(row["tags"]), "scopes": json.loads(row["scopes"]), "owner": row["owner"], "content": row["content"], "resource_ids": json.loads(row["resource_ids"]), "source": json.loads(row["source"]), "eval": json.loads(row["eval"]), "created_at": row["created_at"], "updated_at": row["updated_at"] } finally: conn.close() async def _evolve_knowledge_with_llm(old_content: str, feedback: str) -> str: """使用 LLM 进行知识进化重写""" prompt = f"""你是一个 AI Agent 知识库管理员。请根据反馈建议,对现有的知识内容进行重写进化。 【原知识内容】: {old_content} 【实战反馈建议】: {feedback} 【重写要求】: 1. 融合知识:将反馈中的避坑指南、新参数或修正后的选择逻辑融入原知识,使其更具通用性和准确性。 2. 保持结构:如果原内容有特定格式(如 Markdown、代码示例等),请保持该格式。 3. 语言:简洁直接,使用中文。 4. 禁止:严禁输出任何开场白、解释语或额外的 Markdown 标题,直接返回重写后的正文。 """ try: response = await openrouter_llm_call( messages=[{"role": "user", "content": prompt}], model="google/gemini-2.5-flash-lite" ) evolved = response.get("content", "").strip() if len(evolved) < 5: raise ValueError("LLM output too short") return evolved except Exception as e: print(f"知识进化失败,采用追加模式回退: {e}") return f"{old_content}\n\n---\n[Update {datetime.now().strftime('%Y-%m-%d')}]: {feedback}" @app.put("/api/knowledge/{knowledge_id}") async def update_knowledge(knowledge_id: str, update: KnowledgeUpdateIn): """更新知识评估,支持知识进化""" conn = get_db() try: row = conn.execute("SELECT * FROM knowledge WHERE id = ?", (knowledge_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}") now = datetime.now(timezone.utc).isoformat() eval_data = json.loads(row["eval"]) # 更新评分 if update.update_score is not None: eval_data["score"] = update.update_score # 添加有效案例 if update.add_helpful_case: eval_data["helpful"] = eval_data.get("helpful", 0) + 1 if "helpful_history" not in eval_data: eval_data["helpful_history"] = [] eval_data["helpful_history"].append(update.add_helpful_case) # 添加有害案例 if update.add_harmful_case: eval_data["harmful"] = eval_data.get("harmful", 0) + 1 if "harmful_history" not in eval_data: eval_data["harmful_history"] = [] eval_data["harmful_history"].append(update.add_harmful_case) # 知识进化 content = row["content"] if update.evolve_feedback: content = await _evolve_knowledge_with_llm(content, update.evolve_feedback) eval_data["helpful"] = eval_data.get("helpful", 0) + 1 # 更新数据库 conn.execute( "UPDATE knowledge SET content = ?, eval = ?, updated_at = ? WHERE id = ?", (content, json.dumps(eval_data, ensure_ascii=False), now, knowledge_id) ) conn.commit() return {"status": "ok", "knowledge_id": knowledge_id} finally: conn.close() @app.patch("/api/knowledge/{knowledge_id}") def patch_knowledge(knowledge_id: str, patch: KnowledgePatchIn): """直接编辑知识字段""" conn = get_db() try: row = conn.execute("SELECT * FROM knowledge WHERE id = ?", (knowledge_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}") updates = [] params = [] if patch.task is not None: updates.append("task = ?") params.append(patch.task) if patch.content is not None: updates.append("content = ?") params.append(patch.content) if patch.types is not None: updates.append("types = ?") params.append(json.dumps(patch.types, ensure_ascii=False)) if patch.tags is not None: updates.append("tags = ?") params.append(json.dumps(patch.tags, ensure_ascii=False)) if patch.scopes is not None: updates.append("scopes = ?") params.append(json.dumps(patch.scopes, ensure_ascii=False)) if patch.owner is not None: updates.append("owner = ?") params.append(patch.owner) if not updates: return {"status": "ok", "knowledge_id": knowledge_id} now = datetime.now(timezone.utc).isoformat() updates.append("updated_at = ?") params.append(now) params.append(knowledge_id) query = f"UPDATE knowledge SET {', '.join(updates)} WHERE id = ?" conn.execute(query, params) conn.commit() return {"status": "ok", "knowledge_id": knowledge_id} finally: conn.close() @app.post("/api/knowledge/batch_update") async def batch_update_knowledge(batch: KnowledgeBatchUpdateIn): """批量反馈知识有效性""" if not batch.feedback_list: return {"status": "ok", "updated": 0} conn = get_db() try: # 先处理无需进化的,收集需要进化的 evolution_tasks = [] # [(knowledge_id, old_content, feedback, eval_data)] simple_updates = [] # [(knowledge_id, is_effective, eval_data)] for item in batch.feedback_list: knowledge_id = item.get("knowledge_id") is_effective = item.get("is_effective") feedback = item.get("feedback", "") if not knowledge_id: continue row = conn.execute("SELECT * FROM knowledge WHERE id = ?", (knowledge_id,)).fetchone() if not row: continue eval_data = json.loads(row["eval"]) if is_effective and feedback: evolution_tasks.append((knowledge_id, row["content"], feedback, eval_data)) else: simple_updates.append((knowledge_id, is_effective, eval_data)) # 执行简单更新 now = datetime.now(timezone.utc).isoformat() for knowledge_id, is_effective, eval_data in simple_updates: if is_effective: eval_data["helpful"] = eval_data.get("helpful", 0) + 1 else: eval_data["harmful"] = eval_data.get("harmful", 0) + 1 conn.execute( "UPDATE knowledge SET eval = ?, updated_at = ? WHERE id = ?", (json.dumps(eval_data, ensure_ascii=False), now, knowledge_id) ) # 并发执行知识进化 if evolution_tasks: print(f"🧬 并发处理 {len(evolution_tasks)} 条知识进化...") evolved_results = await asyncio.gather( *[_evolve_knowledge_with_llm(old, fb) for _, old, fb, _ in evolution_tasks] ) for (knowledge_id, _, _, eval_data), evolved_content in zip(evolution_tasks, evolved_results): eval_data["helpful"] = eval_data.get("helpful", 0) + 1 conn.execute( "UPDATE knowledge SET content = ?, eval = ?, updated_at = ? WHERE id = ?", (evolved_content, json.dumps(eval_data, ensure_ascii=False), now, knowledge_id) ) conn.commit() return {"status": "ok", "updated": len(simple_updates) + len(evolution_tasks)} finally: conn.close() @app.post("/api/knowledge/slim") async def slim_knowledge(model: str = "google/gemini-2.5-flash-lite"): """知识库瘦身:合并语义相似知识""" conn = get_db() try: rows = conn.execute("SELECT * FROM knowledge").fetchall() if len(rows) < 2: return {"status": "ok", "message": f"知识库仅有 {len(rows)} 条,无需瘦身"} # 构造发给大模型的内容 entries_text = "" for row in rows: eval_data = json.loads(row["eval"]) types = json.loads(row["types"]) entries_text += f"[ID: {row['id']}] [Types: {','.join(types)}] " entries_text += f"[Helpful: {eval_data.get('helpful', 0)}, Harmful: {eval_data.get('harmful', 0)}] [Score: {eval_data.get('score', 3)}]\n" entries_text += f"Task: {row['task']}\n" entries_text += f"Content: {row['content'][:200]}...\n\n" prompt = f"""你是一个 AI Agent 知识库管理员。以下是当前知识库的全部条目,请执行瘦身操作: 【任务】: 1. 识别语义高度相似或重复的知识,将它们合并为一条更精炼、更通用的知识。 2. 合并时保留 helpful 最高的那条的 ID(helpful 取各条之和)。 3. 对于独立的、无重复的知识,保持原样不动。 【当前知识库】: {entries_text} 【输出格式要求】: 严格按以下格式输出每条知识,条目之间用 === 分隔: ID: <保留的id> TYPES: <逗号分隔的type列表> HELPFUL: <合并后的helpful计数> HARMFUL: <合并后的harmful计数> SCORE: <评分> TASK: <任务描述> CONTENT: <合并后的知识内容> === 最后输出合并报告: REPORT: 原有 X 条,合并后 Y 条,精简了 Z 条。 禁止输出任何开场白或解释。""" print(f"\n[知识瘦身] 正在调用 {model} 分析 {len(rows)} 条知识...") response = await openrouter_llm_call( messages=[{"role": "user", "content": prompt}], model=model ) content = response.get("content", "").strip() if not content: raise HTTPException(status_code=500, detail="LLM 返回为空") # 解析大模型输出 report_line = "" new_entries = [] blocks = [b.strip() for b in content.split("===") if b.strip()] for block in blocks: if block.startswith("REPORT:"): report_line = block continue lines = block.split("\n") kid, types, helpful, harmful, score, task, content_lines = None, [], 0, 0, 3, "", [] current_field = None for line in lines: if line.startswith("ID:"): kid = line[3:].strip() current_field = None elif line.startswith("TYPES:"): types_str = line[6:].strip() types = [t.strip() for t in types_str.split(",") if t.strip()] current_field = None elif line.startswith("HELPFUL:"): try: helpful = int(line[8:].strip()) except Exception: helpful = 0 current_field = None elif line.startswith("HARMFUL:"): try: harmful = int(line[8:].strip()) except Exception: harmful = 0 current_field = None elif line.startswith("SCORE:"): try: score = int(line[6:].strip()) except Exception: score = 3 current_field = None elif line.startswith("TASK:"): task = line[5:].strip() current_field = "task" elif line.startswith("CONTENT:"): content_lines.append(line[8:].strip()) current_field = "content" elif current_field == "task": task += "\n" + line elif current_field == "content": content_lines.append(line) if kid and content_lines: new_entries.append({ "id": kid, "types": types if types else ["strategy"], "helpful": helpful, "harmful": harmful, "score": score, "task": task.strip(), "content": "\n".join(content_lines).strip() }) if not new_entries: raise HTTPException(status_code=500, detail="解析大模型输出失败") # 原子化写回 now = datetime.now(timezone.utc).isoformat() conn.execute("DELETE FROM knowledge") for e in new_entries: eval_data = { "score": e["score"], "helpful": e["helpful"], "harmful": e["harmful"], "confidence": 0.9, "helpful_history": [], "harmful_history": [] } source = { "name": "slim", "category": "exp", "urls": [], "agent_id": "slim", "submitted_by": "system", "timestamp": now } conn.execute( """INSERT INTO knowledge (id, message_id, types, task, tags, scopes, owner, content, source, eval, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( e["id"], "", json.dumps(e["types"]), e["task"], json.dumps({}), json.dumps(["org:cybertogether"]), "agent:slim", e["content"], json.dumps(source, ensure_ascii=False), json.dumps(eval_data, ensure_ascii=False), now, now ) ) conn.commit() result_msg = f"瘦身完成:{len(rows)} → {len(new_entries)} 条知识" if report_line: result_msg += f"\n{report_line}" print(f"[知识瘦身] {result_msg}") return {"status": "ok", "before": len(rows), "after": len(new_entries), "report": report_line} finally: conn.close() @app.post("/api/extract") async def extract_knowledge_from_messages(extract_req: MessageExtractIn): """从消息历史中提取知识(LLM 分析)""" if not extract_req.submitted_by: raise HTTPException(status_code=400, detail="submitted_by is required") messages = extract_req.messages if not messages or len(messages) == 0: return {"status": "ok", "extracted_count": 0, "knowledge_ids": []} # 构造消息历史文本 messages_text = "" for msg in messages: role = msg.get("role", "unknown") content = msg.get("content", "") messages_text += f"[{role}]: {content}\n\n" # LLM 提取知识 prompt = f"""你是一个知识提取专家。请从以下 Agent 对话历史中提取有价值的知识。 【对话历史】: {messages_text} 【提取要求】: 1. 识别对话中的关键知识点(工具使用经验、问题解决方案、最佳实践、踩坑经验等) 2. 每条知识必须包含: - task: 任务场景描述(在什么情况下,要完成什么目标) - content: 核心知识内容(具体可操作的方法、注意事项) - types: 知识类型(从 strategy/tool/user_profile/usecase/definition/plan 中选择) - score: 评分 1-5(根据知识的价值和可操作性) 3. 只提取有实际价值的知识,不要提取泛泛而谈的内容,一次就成功或比较简单的经验就不要记录了。 4. 如果没有值得提取的知识,返回空列表 【输出格式】: 严格按以下 JSON 格式输出,每条知识之间用逗号分隔: [ {{ "task": "任务场景描述", "content": "核心知识内容", "types": ["strategy"], "score": 4 }}, {{ "task": "另一个任务场景", "content": "另一个知识内容", "types": ["tool"], "score": 5 }} ] 如果没有知识,输出: [] **注意**:只记录经过多次尝试、或经过用户指导才成功的知识,一次就成功或比较简单的经验就不要记录了。 禁止输出任何解释或额外文本,只输出 JSON 数组。""" try: print(f"\n[Extract] 正在从 {len(messages)} 条消息中提取知识...") response = await openrouter_llm_call( messages=[{"role": "user", "content": prompt}], model="google/gemini-2.5-flash-lite" ) content = response.get("content", "").strip() # 尝试解析 JSON # 移除可能的 markdown 代码块标记 if content.startswith("```json"): content = content[7:] if content.startswith("```"): content = content[3:] if content.endswith("```"): content = content[:-3] content = content.strip() extracted_knowledge = json.loads(content) if not isinstance(extracted_knowledge, list): raise ValueError("LLM output is not a list") # 保存提取的知识 conn = get_db() knowledge_ids = [] now = datetime.now(timezone.utc).isoformat() try: for item in extracted_knowledge: task = item.get("task", "") knowledge_content = item.get("content", "") types = item.get("types", ["strategy"]) score = item.get("score", 3) if not task or not knowledge_content: continue # 生成 ID import uuid timestamp = datetime.now().strftime('%Y%m%d-%H%M%S') random_suffix = uuid.uuid4().hex[:4] knowledge_id = f"knowledge-{timestamp}-{random_suffix}" # 准备数据 source = { "name": "message_extraction", "category": "exp", "urls": [], "agent_id": extract_req.agent_id, "submitted_by": extract_req.submitted_by, "timestamp": now, "session_key": extract_req.session_key } eval_data = { "score": score, "helpful": 1, "harmful": 0, "confidence": 0.7, "helpful_history": [], "harmful_history": [] } # 插入数据库 conn.execute( """INSERT INTO knowledge (id, message_id, types, task, tags, scopes, owner, content, source, eval, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( knowledge_id, "", json.dumps(types), task, json.dumps({}), json.dumps(["org:cybertogether"]), extract_req.submitted_by, knowledge_content, json.dumps(source, ensure_ascii=False), json.dumps(eval_data, ensure_ascii=False), now, now, ), ) knowledge_ids.append(knowledge_id) conn.commit() print(f"[Extract] 成功提取并保存 {len(knowledge_ids)} 条知识") return { "status": "ok", "extracted_count": len(knowledge_ids), "knowledge_ids": knowledge_ids } finally: conn.close() except json.JSONDecodeError as e: print(f"[Extract] JSON 解析失败: {e}") print(f"[Extract] LLM 输出: {content[:500]}") return {"status": "error", "error": "Failed to parse LLM output", "extracted_count": 0} except Exception as e: print(f"[Extract] 提取失败: {e}") return {"status": "error", "error": str(e), "extracted_count": 0} @app.get("/", response_class=HTMLResponse) def frontend(): """KnowHub 管理前端""" return """