Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/graph_maintenance.py:266-468 (handler)

  The core implementation of the rebuild_communities tool. It clears existing communities for the given level, constructs a weighted undirected graph from paper-entity relations (filtering nodes by min_df), applies Leiden clustering with specified resolution, and persists the new communities and their members to the database. Returns list of new communities with top entities.

  def rebuild_communities(
      level: str = "macro",
      min_df: int = 3,
      resolution: float = 1.0,
  ) -> dict[str, Any]:
      """重建社区
      
      清除指定层级的旧社区并重新构建。
      
      Args:
          level: 社区层级，"macro" 或 "micro"
          min_df: 节点最小文档频率，默认 3
          resolution: Leiden 分辨率参数，默认 1.0
          
      Returns:
          新社区列表
      """
      try:
          # 直接调用 build_communities_v1 with rebuild=True
          from paperlib_mcp.tools.graph_community import register_graph_community_tools
          
          # 由于我们需要直接调用逻辑，这里重新实现简化版本
          # 或者导入并使用内部逻辑
          
          try:
              import igraph as ig
              import leidenalg
          except ImportError:
              return {
                  "communities": [],
                  "error": {
                      "code": "DEPENDENCY_ERROR",
                      "message": "igraph and leidenalg are required"
                  }
              }
          
          from paperlib_mcp.tools.graph_community import (
              COMMUNITY_ENTITY_TYPES,
              get_edge_weight,
          )
          from paperlib_mcp.models_graph import CommunityBrief
          from collections import defaultdict
          import json
          
          with get_db() as conn:
              # 清除旧结果
              with conn.cursor() as cur:
                  cur.execute(
                      """
                      DELETE FROM community_members 
                      WHERE comm_id IN (SELECT comm_id FROM communities WHERE level = %s)
                      """,
                      (level,)
                  )
                  cur.execute(
                      """
                      DELETE FROM community_summaries 
                      WHERE comm_id IN (SELECT comm_id FROM communities WHERE level = %s)
                      """,
                      (level,)
                  )
                  cur.execute(
                      "DELETE FROM communities WHERE level = %s",
                      (level,)
                  )
              
              # 获取 Paper->Entity 关系
              relations = query_all(
                  """
                  SELECT 
                      p.entity_id AS paper_eid,
                      p.canonical_key AS doc_id,
                      x.entity_id AS node_eid,
                      x.type AS node_type,
                      x.canonical_name
                  FROM relations r
                  JOIN entities p ON p.entity_id = r.subj_entity_id AND p.type = 'Paper'
                  JOIN entities x ON x.entity_id = r.obj_entity_id
                  WHERE r.predicate IN (
                      'PAPER_HAS_TOPIC', 'PAPER_USES_MEASURE', 'PAPER_IDENTIFIES_WITH',
                      'PAPER_IN_SETTING', 'PAPER_USES_DATA'
                  )
                  AND x.type = ANY(%s)
                  """,
                  (COMMUNITY_ENTITY_TYPES,)
              )
              
              if not relations:
                  return {"communities": [], "error": {"code": "NOT_FOUND", "message": "No relations found"}}
              
              # 构建图并聚类（同 build_communities_v1）
              paper_to_nodes: dict[str, set[int]] = defaultdict(set)
              node_info: dict[int, dict] = {}
              
              for r in relations:
                  paper_to_nodes[r["doc_id"]].add(r["node_eid"])
                  if r["node_eid"] not in node_info:
                      node_info[r["node_eid"]] = {
                          "entity_id": r["node_eid"],
                          "type": r["node_type"],
                          "canonical_name": r["canonical_name"],
                      }
              
              node_df: dict[int, int] = defaultdict(int)
              for doc_id, nodes in paper_to_nodes.items():
                  for node_id in nodes:
                      node_df[node_id] += 1
              
              valid_nodes = {nid for nid, df in node_df.items() if df >= min_df}
              
              if not valid_nodes:
                  return {"communities": [], "error": {"code": "NOT_FOUND", "message": f"No nodes with df >= {min_df}"}}
              
              # 构建边
              edge_counts: dict[tuple[int, int], float] = defaultdict(float)
              for doc_id, nodes in paper_to_nodes.items():
                  valid_doc_nodes = [n for n in nodes if n in valid_nodes]
                  for i, n1 in enumerate(valid_doc_nodes):
                      for n2 in valid_doc_nodes[i+1:]:
                          key = (min(n1, n2), max(n1, n2))
                          weight = get_edge_weight(node_info[n1]["type"], node_info[n2]["type"])
                          edge_counts[key] += weight
              
              if not edge_counts:
                  return {"communities": [], "error": {"code": "NOT_FOUND", "message": "No edges found"}}
              
              # 构建 igraph
              node_list = sorted(valid_nodes)
              node_to_idx = {nid: idx for idx, nid in enumerate(node_list)}
              
              edges = []
              weights = []
              for (n1, n2), w in edge_counts.items():
                  edges.append((node_to_idx[n1], node_to_idx[n2]))
                  weights.append(w)
              
              g = ig.Graph(n=len(node_list), edges=edges, directed=False)
              g.es["weight"] = weights
              
              # Leiden 聚类
              partition = leidenalg.find_partition(
                  g, leidenalg.RBConfigurationVertexPartition,
                  weights="weight", resolution_parameter=resolution
              )
              
              # 写入数据库
              communities_result = []
              community_members_map: dict[int, list[tuple[int, float]]] = defaultdict(list)
              
              for node_idx, comm_idx in enumerate(partition.membership):
                  node_id = node_list[node_idx]
                  weight = float(node_df[node_id])
                  community_members_map[comm_idx].append((node_id, weight))
              
              with conn.cursor() as cur:
                  for comm_idx, members in community_members_map.items():
                      if len(members) < 2:
                          continue
                      
                      cur.execute(
                          """
                          INSERT INTO communities(level, method, params)
                          VALUES (%s, 'leiden', %s::jsonb)
                          RETURNING comm_id
                          """,
                          (level, json.dumps({"resolution": resolution, "min_df": min_df}))
                      )
                      result = cur.fetchone()
                      comm_id = result["comm_id"]
                      
                      for node_id, weight in members:
                          cur.execute(
                              """
                              INSERT INTO community_members(comm_id, entity_id, role, weight)
                              VALUES (%s, %s, 'member', %s)
                              """,
                              (comm_id, node_id, weight)
                          )
                      
                      sorted_members = sorted(members, key=lambda x: x[1], reverse=True)
                      top_entities = []
                      for node_id, weight in sorted_members[:20]:
                          info = node_info.get(node_id, {})
                          top_entities.append({
                              "entity_id": node_id,
                              "type": info.get("type", ""),
                              "canonical_name": info.get("canonical_name", ""),
                              "weight": weight,
                          })
                      
                      communities_result.append({
                          "comm_id": comm_id,
                          "size": len(members),
                          "top_entities": top_entities,
                      })
              
              communities_result.sort(key=lambda x: x["size"], reverse=True)
              
              return {"communities": communities_result}
              
      except Exception as e:
          return {"communities": [], "error": {"code": "DB_CONN_ERROR", "message": str(e)}}

• src/paperlib_mcp/server.py:44-44 (registration)

  The call that registers all graph maintenance tools (including rebuild_communities) to the main FastMCP server instance.

  register_graph_maintenance_tools(mcp)

• src/paperlib_mcp/tools/graph_maintenance.py:23-24 (registration)

  Function that defines and registers the graph maintenance tools using @mcp.tool() decorators when called with an MCP instance.

  def register_graph_maintenance_tools(mcp: FastMCP) -> None:
      """注册 GraphRAG 维护工具"""