Paperlib MCP

Instructions

Implementation Reference

• src/paperlib_mcp/tools/review.py:284-502 (handler)

  The core handler function for the 'build_section_evidence_pack_v1' tool. It builds a reproducible evidence pack for a specific section by selecting relevant chunks based on entity types, claims, or keywords from the database, applying limits, and storing in evidence_packs table.

  @mcp.tool()
  def build_section_evidence_pack_v1(
      outline_id: str,
      section_id: str,
      max_chunks: int = 60,
      per_doc_limit: int = 4,
      rebuild: bool = False,
  ) -> dict[str, Any]:
      """构建章节证据包
  
      为指定章节生成固定的证据包（可复现）。
  
      Args:
          outline_id: 大纲 ID
          section_id: 章节 ID
          max_chunks: 最大 chunk 数量，默认 60
          per_doc_limit: 每篇文档最多 chunk 数，默认 4
          rebuild: 是否重建，默认 False
  
      Returns:
          pack_id, chunk_count, doc_count
      """
      try:
          # 检查缓存
          if not rebuild:
              cached = query_one(
                  """
                  SELECT pack_id, params, created_at::text
                  FROM review_section_packs
                  WHERE outline_id = %s AND section_id = %s
                  """,
                  (outline_id, section_id),
              )
              if cached:
                  # 获取 pack 统计
                  stats = query_one(
                      """
                      SELECT COUNT(*) as chunk_count, COUNT(DISTINCT doc_id) as doc_count
                      FROM evidence_pack_items
                      WHERE pack_id = %s
                      """,
                      (cached["pack_id"],),
                  )
                  return {
                      "pack_id": cached["pack_id"],
                      "section_id": section_id,
                      "chunk_count": stats["chunk_count"] if stats else 0,
                      "doc_count": stats["doc_count"] if stats else 0,
                      "params": cached["params"],
                      "created_at": cached["created_at"],
                      "reused": True,
                  }
  
          # 获取 outline 和 section 信息
          outline = query_one(
              "SELECT topic, sources_json FROM review_outlines WHERE outline_id = %s",
              (outline_id,),
          )
          if not outline:
              return {"error": f"Outline not found: {outline_id}"}
  
          section = query_one(
              """
              SELECT section_id, title, sources_json, keywords
              FROM review_outline_sections
              WHERE outline_id = %s AND section_id = %s
              """,
              (outline_id, section_id),
          )
          if not section:
              return {"error": f"Section not found: {section_id}"}
  
          # 根据 section 类型选择 chunks
          outline_sources = outline["sources_json"] or {}
          section_sources = section["sources_json"] or {}
          entity_types = section_sources.get("entity_types", [])
          keywords = section["keywords"] or []
  
          # 候选 chunk_ids
          candidate_chunks = []
  
          # 策略 1: 从实体类型对应的 mentions 获取 chunks
          if entity_types:
              entity_chunks = query_all(
                  """
                  SELECT DISTINCT m.chunk_id, m.doc_id, c.page_start, c.page_end
                  FROM mentions m
                  JOIN entities e ON m.entity_id = e.entity_id
                  JOIN chunks c ON m.chunk_id = c.chunk_id
                  WHERE e.type = ANY(%s)
                  ORDER BY m.chunk_id
                  LIMIT %s
                  """,
                  (entity_types, max_chunks * 2),
              )
              candidate_chunks.extend(entity_chunks)
  
          # 策略 2: 对于 findings/debates，使用 claims
          if section_id in ("findings", "debates"):
              if section_id == "findings":
                  # 高置信度 claims - 使用子查询避免 DISTINCT + ORDER BY 冲突
                  claim_chunks = query_all(
                      """
                      SELECT chunk_id, doc_id, page_start, page_end FROM (
                          SELECT DISTINCT ON (c.chunk_id) 
                              c.chunk_id, c.doc_id, ch.page_start, ch.page_end, c.confidence
                          FROM claims c
                          JOIN chunks ch ON c.chunk_id = ch.chunk_id
                          WHERE c.confidence >= 0.7
                          ORDER BY c.chunk_id, c.confidence DESC
                      ) sub
                      ORDER BY confidence DESC
                      LIMIT %s
                      """,
                      (max_chunks * 2,),
                  )
              else:  # debates - 找冲突 sign
                  claim_chunks = query_all(
                      """
                      SELECT DISTINCT c.chunk_id, c.doc_id, ch.page_start, ch.page_end
                      FROM claims c
                      JOIN chunks ch ON c.chunk_id = ch.chunk_id
                      WHERE c.sign IN ('positive', 'negative', 'mixed')
                      ORDER BY c.chunk_id
                      LIMIT %s
                      """,
                      (max_chunks * 2,),
                  )
              candidate_chunks.extend(claim_chunks)
  
          # 策略 3: 使用关键词搜索
          if keywords and len(candidate_chunks) < max_chunks:
              keyword_pattern = "|".join(keywords)
              keyword_chunks = query_all(
                  """
                  SELECT chunk_id, doc_id, page_start, page_end
                  FROM chunks
                  WHERE text ~* %s
                  LIMIT %s
                  """,
                  (keyword_pattern, max_chunks * 2),
              )
              candidate_chunks.extend(keyword_chunks)
  
          # 去重
          seen = set()
          unique_chunks = []
          for chunk in candidate_chunks:
              if chunk["chunk_id"] not in seen:
                  seen.add(chunk["chunk_id"])
                  unique_chunks.append(chunk)
  
          # 应用 per_doc_limit
          doc_counts: dict[str, int] = defaultdict(int)
          filtered_chunks = []
          for chunk in unique_chunks:
              doc_id = chunk["doc_id"]
              if doc_counts[doc_id] < per_doc_limit:
                  filtered_chunks.append(chunk)
                  doc_counts[doc_id] += 1
              if len(filtered_chunks) >= max_chunks:
                  break
  
          # 创建 evidence pack
          params = {
              "max_chunks": max_chunks,
              "per_doc_limit": per_doc_limit,
              "section_id": section_id,
          }
  
          with get_db() as conn:
              with conn.cursor() as cur:
                  # 创建 pack
                  cur.execute(
                      """
                      INSERT INTO evidence_packs (query, params_json)
                      VALUES (%s, %s)
                      RETURNING pack_id
                      """,
                      (f"section:{section_id}:{outline_id}", json.dumps(params)),
                  )
                  pack_id = cur.fetchone()["pack_id"]
  
                  # 插入 items
                  for rank, chunk in enumerate(filtered_chunks):
                      cur.execute(
                          """
                          INSERT INTO evidence_pack_items (pack_id, doc_id, chunk_id, rank)
                          VALUES (%s, %s, %s, %s)
                          """,
                          (pack_id, chunk["doc_id"], chunk["chunk_id"], rank),
                      )
  
                  # 删除旧缓存
                  cur.execute(
                      "DELETE FROM review_section_packs WHERE outline_id = %s AND section_id = %s",
                      (outline_id, section_id),
                  )
  
                  # 缓存新映射
                  cur.execute(
                      """
                      INSERT INTO review_section_packs (outline_id, section_id, pack_id, params)
                      VALUES (%s, %s, %s, %s)
                      """,
                      (outline_id, section_id, pack_id, json.dumps(params)),
                  )
  
          return {
              "pack_id": pack_id,
              "section_id": section_id,
              "chunk_count": len(filtered_chunks),
              "doc_count": len(doc_counts),
              "params": params,
              "reused": False,
          }
  
      except Exception as e:
          return {"error": str(e)}

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

  The registration point where review tools, including 'build_section_evidence_pack_v1', are registered to the MCP server instance by calling register_review_tools(mcp). The actual tool definitions with @mcp.tool() decorators are inside register_review_tools in review.py.

  register_review_tools(mcp)

• src/paperlib_mcp/tools/review.py:104-106 (registration)

  The function that defines and registers all review tools using @mcp.tool() decorators when called.

  def register_review_tools(mcp: FastMCP) -> None:
      """注册 M3 综述工具"""