Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/graph_claim_grouping.py:38-177 (handler)

  The core handler function for build_claim_groups_v1 tool. It queries claims from DB, groups them by composite key (topic|outcome_family|treatment_family|sign|id_family|setting), optionally dry_run previews, or inserts into claim_groups and claim_group_members tables.
  @mcp.tool() def build_claim_groups_v1( scope: str = "all", max_claims_per_doc: int = 100, # v1.1: raised from 20 to improve coverage dry_run: bool = False, ) -> dict[str, Any]: """对结论进行分组/聚类。 Args: scope: 处理范围，"all", "comm_id:...", "doc_ids:id1,id2" max_claims_per_doc: 每个文档最多处理多少条结论 dry_run: 是否仅预览 """ try: # 1. 确定范围 params = [] where_clauses = [] if scope.startswith("comm_id:"): comm_id = int(scope.split(":", 1)[1]) where_clauses.append(""" EXISTS ( SELECT 1 FROM community_members cm JOIN mentions m ON m.entity_id = cm.entity_id WHERE cm.comm_id = %s AND m.doc_id = c.doc_id ) """) params.append(comm_id) elif scope.startswith("doc_ids:"): doc_ids = scope.split(":", 1)[1].split(",") where_clauses.append("c.doc_id = ANY(%s)") params.append(doc_ids) where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" # 2. 获取结论及关联信息 sql = f""" SELECT c.claim_id, c.doc_id, c.claim_text, c.sign, c.conditions, (SELECT e.canonical_key FROM relations r JOIN entities e ON e.entity_id = r.obj_entity_id WHERE r.subj_entity_id = (SELECT entity_id FROM entities WHERE type='Paper' AND canonical_key = c.doc_id) AND r.predicate = 'PAPER_HAS_TOPIC' LIMIT 1) as topic_key, (SELECT e.canonical_key FROM relations r JOIN entities e ON e.entity_id = r.obj_entity_id WHERE r.subj_entity_id = (SELECT entity_id FROM entities WHERE type='Paper' AND canonical_key = c.doc_id) AND r.predicate = 'PAPER_IDENTIFIES_WITH' LIMIT 1) as id_family FROM claims c {where_sql} ORDER BY c.doc_id, c.confidence DESC """ claims = query_all(sql, tuple(params)) if not claims: return BuildClaimGroupsOut(new_groups=0, total_members=0).model_dump() # 3. 分组逻辑 groups: dict[str, list[int]] = defaultdict(list) group_details: dict[str, dict] = {} doc_counts: dict[str, int] = defaultdict(int) for c in claims: if doc_counts[c["doc_id"]] >= max_claims_per_doc: continue doc_counts[c["doc_id"]] += 1 # 构造 group_key topic_key = c["topic_key"] or "unknown_topic" sign = c["sign"] or "null" id_family = c["id_family"] or "general" # 尝试从 conditions 或 claim_text 提取 outcome/treatment conditions = c["conditions"] or {} outcome_family = conditions.get("outcome_family") or extract_family(c["claim_text"], "outcome_gen") treatment_family = conditions.get("treatment_family") or extract_family(c["claim_text"], "treatment_gen") setting = conditions.get("setting") or "general" group_key = f"{topic_key}|{outcome_family}|{treatment_family}|{sign}|{id_family}|{setting}" groups[group_key].append(c["claim_id"]) if group_key not in group_details: # 查找 topic_entity_id topic_ent = query_one("SELECT entity_id FROM entities WHERE canonical_key = %s", (topic_key,)) group_details[group_key] = { "topic_entity_id": topic_ent["entity_id"] if topic_ent else None, "sign": sign, "setting": setting, "id_family": id_family } if dry_run: return BuildClaimGroupsOut( new_groups=len(groups), total_members=sum(len(v) for v in groups.values()) ).model_dump() # 4. 写入数据库 total_members = 0 with get_db() as conn: for key, claim_ids in groups.items(): details = group_details[key] try: with conn.cursor() as cur: with conn.transaction(): cur.execute(""" INSERT INTO claim_groups (group_key, topic_entity_id, sign, setting, id_family) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (group_key) DO UPDATE SET updated_at = now() RETURNING group_id """, (key, details["topic_entity_id"], details["sign"], details["setting"], details["id_family"])) group_id = cur.fetchone()["group_id"] for cid in claim_ids: cur.execute(""" INSERT INTO claim_group_members (group_id, claim_id) VALUES (%s, %s) ON CONFLICT (group_id, claim_id) DO NOTHING """, (group_id, cid)) total_members += cur.rowcount except Exception as e: print(f"Error processing group {key}: {e}") # 计算总数 (在 commit 前查询) with conn.cursor() as cur: cur.execute("SELECT COUNT(*) as count FROM claim_groups") new_groups_total = cur.fetchone()["count"] return BuildClaimGroupsOut( new_groups=new_groups_total, total_members=total_members ).model_dump() except Exception as e: return BuildClaimGroupsOut( new_groups=0, total_members=0, error=MCPErrorModel(code="SYSTEM_ERROR", message=str(e)) ).model_dump()
• src/paperlib_mcp/models_graph.py:430-442 (schema)

  Pydantic models defining the input parameters (scope, max_claims_per_doc, dry_run) and output structure (new_groups, total_members, error) for the build_claim_groups_v1 tool.
  class BuildClaimGroupsIn(BaseModel): """build_claim_groups_v1 输入""" scope: Literal["all"] | str = "all" max_claims_per_doc: int = 20 dry_run: bool = False class BuildClaimGroupsOut(BaseModel): """build_claim_groups_v1 输出""" new_groups: int total_members: int error: Optional[MCPErrorModel] = None
• src/paperlib_mcp/server.py:51-51 (registration)

  Invocation of the registration function that adds the build_claim_groups_v1 tool (via @mcp.tool() decorators) to the FastMCP server instance.
  register_graph_claim_grouping_tools(mcp)
• src/paperlib_mcp/tools/graph_claim_grouping.py:21-32 (helper)

  Utility function used in grouping to extract family categories (earnings, esg, innovation, governance) from claim text or conditions.
  def extract_family(text: str, default: str = "general") -> str: """从文本中提取家族（简单规则）""" text = text.lower() if any(k in text for k in ["earnings", "profit", "accrual"]): return "earnings" if any(k in text for k in ["esg", "csr", "environment", "social"]): return "esg" if any(k in text for k in ["innovation", "r&d", "patent"]): return "innovation" if any(k in text for k in ["governance", "board", "ceo"]): return "governance" return default