rlm_auto_analyze

Instructions

Implementation Reference

• src/rlm_mcp_server.py:1761-1835 (handler)

  The main handler function for the rlm_auto_analyze tool. It auto-detects content type, selects chunking strategy, adapts the query for the goal, and runs batch sub-queries on the chunked content.

  @mcp.tool()
  async def rlm_auto_analyze(
      name: str,
      content: str,
      goal: str,
      provider: str = "auto",
      concurrency: int = 4,
  ) -> dict:
      """Automatically detect content type and analyze with optimal chunking strategy.
  
      One-step analysis for common tasks.
  
      Args:
          name: Context identifier
          content: The content to analyze
          goal: Analysis goal: 'summarize', 'find_bugs', 'extract_structure', 'security_audit', or 'answer:<your question>'
          provider: LLM provider - 'auto' prefers Ollama if available
          concurrency: Max parallel requests (default 4, max 8)
      """
      concurrency = min(concurrency, 8)
  
      # Load the content (call implementation directly, not the tool)
      await _load_context_impl(name=name, content=content)
  
      # Detect content type
      detection = _detect_content_type(content)
      detected_type = detection["type"]
      confidence = detection["confidence"]
  
      # Select chunking strategy
      strategy_config = _select_chunking_strategy(detected_type)
  
      # Chunk the content (call implementation directly, not the tool)
      chunk_result = await _chunk_context_impl(
          name=name,
          strategy=strategy_config["strategy"],
          size=strategy_config["size"],
      )
      chunk_count = chunk_result["chunk_count"]
  
      # Sample if too many chunks (max 20)
      chunk_indices = list(range(chunk_count))
      sampled = False
      if chunk_count > 20:
          step = max(1, chunk_count // 20)
          chunk_indices = list(range(0, chunk_count, step))[:20]
          sampled = True
  
      # Adapt query for goal and content type
      adapted_query = _adapt_query_for_goal(goal, detected_type)
  
      # Run batch query (call implementation directly, not the tool)
      batch_result = await _sub_query_batch_impl(
          query=adapted_query,
          context_name=name,
          chunk_indices=chunk_indices,
          provider=provider,
          concurrency=concurrency,
      )
  
      return {
          "status": "completed",
          "detected_type": detected_type,
          "confidence": confidence,
          "strategy": strategy_config,
          "chunk_count": chunk_count,
          "chunks_analyzed": len(chunk_indices),
          "sampled": sampled,
          "goal": goal,
          "adapted_query": adapted_query,
          "provider": provider,
          "successful": batch_result["successful"],
          "failed": batch_result["failed"],
          "results": batch_result["results"],
      }

• src/rlm_mcp_server.py:1762-1779 (schema)

  The input schema for rlm_auto_analyze defined by the function signature and docstring parameters: name (str), content (str), goal (str), provider (str, default 'auto'), concurrency (int, default 4).

  async def rlm_auto_analyze(
      name: str,
      content: str,
      goal: str,
      provider: str = "auto",
      concurrency: int = 4,
  ) -> dict:
      """Automatically detect content type and analyze with optimal chunking strategy.
  
      One-step analysis for common tasks.
  
      Args:
          name: Context identifier
          content: The content to analyze
          goal: Analysis goal: 'summarize', 'find_bugs', 'extract_structure', 'security_audit', or 'answer:<your question>'
          provider: LLM provider - 'auto' prefers Ollama if available
          concurrency: Max parallel requests (default 4, max 8)
      """

• src/rlm_mcp_server.py:1761-1761 (registration)

  The tool is registered via the @mcp.tool() decorator on line 1761, using FastMCP's decorator-based registration pattern.

  @mcp.tool()

• src/rlm_mcp_server.py:1153-1163 (helper)

  Helper function _select_chunking_strategy selects the chunking strategy based on detected content type (used by rlm_auto_analyze).

  def _select_chunking_strategy(content_type: str) -> dict:
      """Select chunking strategy based on content type."""
      strategies = {
          "python": {"strategy": "lines", "size": 150},
          "code": {"strategy": "lines", "size": 150},
          "json": {"strategy": "chars", "size": 10000},
          "markdown": {"strategy": "paragraphs", "size": 20},
          "logs": {"strategy": "lines", "size": 500},
          "prose": {"strategy": "paragraphs", "size": 30},
      }
      return strategies.get(content_type, {"strategy": "lines", "size": 100})

• src/rlm_mcp_server.py:1166-1201 (helper)

  Helper function _adapt_query_for_goal generates appropriate sub-queries based on analysis goal and content type (used by rlm_auto_analyze).

  def _adapt_query_for_goal(goal: str, content_type: str) -> str:
      """Generate appropriate sub-query based on goal and content type."""
      if goal.startswith("answer:"):
          return goal[7:].strip()
  
      goal_templates = {
          "find_bugs": {
              "python": "Identify bugs, issues, or potential errors in this Python code. Look for: syntax errors, logic errors, unhandled exceptions, type mismatches, missing imports.",
              "code": "Identify bugs, issues, or potential errors in this code. Look for: syntax errors, logic errors, unhandled exceptions.",
              "default": "Identify any errors, issues, or problems in this content.",
          },
          "summarize": {
              "python": "Summarize what this Python code does. List main functions/classes and their purpose.",
              "code": "Summarize what this code does. List main functions and their purpose.",
              "markdown": "Summarize the main points of this documentation in 2-3 sentences.",
              "prose": "Summarize the main points of this text in 2-3 sentences.",
              "logs": "Summarize the key events and errors in these logs.",
              "json": "Summarize the structure and key data in this JSON.",
              "default": "Summarize the main points of this content in 2-3 sentences.",
          },
          "extract_structure": {
              "python": "Extract the code structure: list all classes, functions, and their signatures.",
              "code": "Extract the code structure: list all functions/classes and their signatures.",
              "json": "Extract the JSON schema: list top-level keys and their types.",
              "markdown": "Extract the document structure: list all headings and hierarchy.",
              "default": "Extract the main structural elements of this content.",
          },
          "security_audit": {
              "python": "Find security vulnerabilities: SQL injection, command injection, eval(), exec(), unsafe deserialization, hardcoded secrets, path traversal.",
              "code": "Find security vulnerabilities: injection flaws, unsafe functions, hardcoded credentials.",
              "default": "Identify potential security issues or sensitive information.",
          },
      }
  
      templates = goal_templates.get(goal, {})
      return templates.get(content_type, templates.get("default", f"Analyze this content for: {goal}"))