""" 知识管理工具 - KnowHub API 封装 所有工具通过 HTTP API 调用 KnowHub Server。 """ import os import logging import httpx from typing import List, Dict, Optional, Any from agent.tools import tool, ToolResult, ToolContext logger = logging.getLogger(__name__) # KnowHub Server API 地址 KNOWHUB_API = os.getenv("KNOWHUB_API", "http://localhost:8000") @tool() async def knowledge_search( query: str, top_k: int = 5, min_score: int = 3, tags_type: Optional[List[str]] = None, context: Optional[ToolContext] = None, ) -> ToolResult: """ 检索知识(两阶段:语义路由 + 质量精排) Args: query: 搜索查询(任务描述) top_k: 返回数量(默认 5) min_score: 最低评分过滤(默认 3) tags_type: 按类型过滤(tool/usecase/definition/plan/strategy) context: 工具上下文 Returns: 相关知识列表 """ try: params = { "q": query, "top_k": top_k, "min_score": min_score, } if tags_type: params["tags_type"] = ",".join(tags_type) async with httpx.AsyncClient(timeout=60.0) as client: response = await client.get(f"{KNOWHUB_API}/api/knowledge/search", params=params) response.raise_for_status() data = response.json() results = data.get("results", []) count = data.get("count", 0) if not results: return ToolResult( title="🔍 未找到相关知识", output=f"查询: {query}\n\n知识库中暂无相关的高质量知识。", long_term_memory=f"知识检索: 未找到相关知识 - {query[:50]}" ) # 格式化输出 output_lines = [f"查询: {query}\n", f"找到 {count} 条相关知识:\n"] for idx, item in enumerate(results, 1): output_lines.append(f"\n### {idx}. [{item['id']}] (⭐ {item.get('score', 3)})") output_lines.append(f"**场景**: {item['scenario'][:150]}...") output_lines.append(f"**内容**: {item['content'][:200]}...") return ToolResult( title="✅ 知识检索成功", output="\n".join(output_lines), long_term_memory=f"知识检索: 找到 {count} 条相关知识 - {query[:50]}", metadata={ "count": count, "knowledge_ids": [item["id"] for item in results], "items": results } ) except Exception as e: logger.error(f"知识检索失败: {e}") return ToolResult( title="❌ 检索失败", output=f"错误: {str(e)}", error=str(e) ) @tool() async def knowledge_save( scenario: str, content: str, tags_type: List[str], urls: List[str] = None, agent_id: str = "research_agent", score: int = 3, message_id: str = "", context: Optional[ToolContext] = None, ) -> ToolResult: """ 保存新知识 Args: scenario: 任务描述(在什么情景下 + 要完成什么目标) content: 核心内容 tags_type: 知识类型标签,可选:tool, usecase, definition, plan, strategy urls: 参考来源链接列表 agent_id: 执行此调研的 agent ID score: 初始评分 1-5(默认 3) message_id: 来源 Message ID context: 工具上下文 Returns: 保存结果 """ try: payload = { "scenario": scenario, "content": content, "tags_type": tags_type, "urls": urls or [], "agent_id": agent_id, "score": score, "message_id": message_id } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(f"{KNOWHUB_API}/api/knowledge", json=payload) response.raise_for_status() data = response.json() knowledge_id = data.get("knowledge_id", "unknown") return ToolResult( title="✅ 知识已保存", output=f"知识 ID: {knowledge_id}\n\n场景:\n{scenario[:100]}...", long_term_memory=f"保存知识: {knowledge_id} - {scenario[:50]}", metadata={"knowledge_id": knowledge_id} ) except Exception as e: logger.error(f"保存知识失败: {e}") return ToolResult( title="❌ 保存失败", output=f"错误: {str(e)}", error=str(e) ) @tool() async def knowledge_update( knowledge_id: str, add_helpful_case: Optional[Dict] = None, add_harmful_case: Optional[Dict] = None, update_score: Optional[int] = None, evolve_feedback: Optional[str] = None, context: Optional[ToolContext] = None, ) -> ToolResult: """ 更新已有知识的评估反馈 Args: knowledge_id: 知识 ID add_helpful_case: 添加好用的案例 add_harmful_case: 添加不好用的案例 update_score: 更新评分(1-5) evolve_feedback: 经验进化反馈(触发 LLM 重写) context: 工具上下文 Returns: 更新结果 """ try: payload = {} if add_helpful_case: payload["add_helpful_case"] = add_helpful_case if add_harmful_case: payload["add_harmful_case"] = add_harmful_case if update_score is not None: payload["update_score"] = update_score if evolve_feedback: payload["evolve_feedback"] = evolve_feedback if not payload: return ToolResult( title="⚠️ 无更新", output="未指定任何更新内容", long_term_memory="尝试更新知识但未指定更新内容" ) async with httpx.AsyncClient(timeout=60.0) as client: response = await client.put(f"{KNOWHUB_API}/api/knowledge/{knowledge_id}", json=payload) response.raise_for_status() summary = [] if add_helpful_case: summary.append("添加 helpful 案例") if add_harmful_case: summary.append("添加 harmful 案例") if update_score is not None: summary.append(f"更新评分: {update_score}") if evolve_feedback: summary.append("知识进化: 基于反馈重写内容") return ToolResult( title="✅ 知识已更新", output=f"知识 ID: {knowledge_id}\n\n更新内容:\n" + "\n".join(f"- {s}" for s in summary), long_term_memory=f"更新知识: {knowledge_id}" ) except Exception as e: logger.error(f"更新知识失败: {e}") return ToolResult( title="❌ 更新失败", output=f"错误: {str(e)}", error=str(e) ) @tool() async def knowledge_batch_update( feedback_list: List[Dict[str, Any]], context: Optional[ToolContext] = None, ) -> ToolResult: """ 批量反馈知识的有效性 Args: feedback_list: 评价列表,每个元素包含: - knowledge_id: (str) 知识 ID - is_effective: (bool) 是否有效 - feedback: (str, optional) 改进建议,若有效且有建议则触发知识进化 context: 工具上下文 Returns: 批量更新结果 """ try: if not feedback_list: return ToolResult( title="⚠️ 反馈列表为空", output="未提供任何反馈", long_term_memory="批量更新知识: 反馈列表为空" ) payload = {"feedback_list": feedback_list} async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post(f"{KNOWHUB_API}/api/knowledge/batch_update", json=payload) response.raise_for_status() data = response.json() updated = data.get("updated", 0) return ToolResult( title="✅ 批量更新完成", output=f"成功更新 {updated} 条知识", long_term_memory=f"批量更新知识: 成功 {updated} 条" ) except Exception as e: logger.error(f"批量更新知识失败: {e}") return ToolResult( title="❌ 批量更新失败", output=f"错误: {str(e)}", error=str(e) ) @tool() async def knowledge_list( limit: int = 10, tags_type: Optional[List[str]] = None, context: Optional[ToolContext] = None, ) -> ToolResult: """ 列出已保存的知识 Args: limit: 返回数量限制(默认 10) tags_type: 按类型过滤(可选) context: 工具上下文 Returns: 知识列表 """ try: params = {"limit": limit} if tags_type: params["tags_type"] = ",".join(tags_type) async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get(f"{KNOWHUB_API}/api/knowledge", params=params) response.raise_for_status() data = response.json() results = data.get("results", []) count = data.get("count", 0) if not results: return ToolResult( title="📂 知识库为空", output="还没有保存任何知识", long_term_memory="知识库为空" ) output_lines = [f"共找到 {count} 条知识:\n"] for item in results: score = item.get("eval", {}).get("score", 3) output_lines.append(f"- [{item['id']}] (⭐{score}) {item['scenario'][:60]}...") return ToolResult( title="📚 知识列表", output="\n".join(output_lines), long_term_memory=f"列出 {count} 条知识" ) except Exception as e: logger.error(f"列出知识失败: {e}") return ToolResult( title="❌ 列表失败", output=f"错误: {str(e)}", error=str(e) ) @tool() async def knowledge_slim( model: str = "anthropic/claude-sonnet-4.5", context: Optional[ToolContext] = None, ) -> ToolResult: """ 知识库瘦身:调用顶级大模型,将知识库中语义相似的知识合并精简 Args: model: 使用的模型(默认 claude-sonnet-4.5) context: 工具上下文 Returns: 瘦身结果报告 """ try: async with httpx.AsyncClient(timeout=300.0) as client: response = await client.post(f"{KNOWHUB_API}/api/knowledge/slim", params={"model": model}) response.raise_for_status() data = response.json() before = data.get("before", 0) after = data.get("after", 0) report = data.get("report", "") result = f"瘦身完成:{before} → {after} 条知识" if report: result += f"\n{report}" return ToolResult( title="✅ 知识库瘦身完成", output=result, long_term_memory=f"知识库瘦身: {before} → {after} 条" ) except Exception as e: logger.error(f"知识库瘦身失败: {e}") return ToolResult( title="❌ 瘦身失败", output=f"错误: {str(e)}", error=str(e) ) # 兼容接口:get_experience @tool(description="检索历史经验(strategy 标签的知识)") async def get_experience( query: str, k: int = 3, context: Optional[ToolContext] = None, ) -> ToolResult: """ 检索历史经验(兼容接口,实际调用 knowledge_search 并过滤 strategy 标签) Args: query: 搜索查询(任务描述) k: 返回数量(默认 3) context: 工具上下文 Returns: 相关经验列表 """ return await knowledge_search( query=query, top_k=k, min_score=1, # 经验的评分门槛较低 tags_type=["strategy"], context=context )