bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

Add duplicate entry prevention to LLM tool pipeline

Browse Source 
Three-layer guard on create_note and create_task:
1. Exact title match (get_note_by_title, case-insensitive) → hard block,
   redirects model to use update_note instead
2. Fuzzy title match (SequenceMatcher ≥82%, punct-stripped word search for
   candidates) → hard block; catches "Game Premise" vs "Game Premise Notes"
3. Semantic content similarity (semantic_search_notes threshold=0.87) → soft
   block with requires_confirmation:true; model asks user then retries with
   confirmed:true; graceful no-op if embedding model is unavailable

create_project and create_milestone now always require confirmed:true before
creation (structural entities — must be intentional). Both are also guarded
by exact + fuzzy title checks post-confirmation.

create_note and create_task gain an optional confirmed parameter (not
required by default; only used to bypass a content-similarity soft-block
after the user has been asked).

Helpers added to tools.py:
- _fuzzy_title_match(title, candidates, threshold=0.82)
- _PUNCT_RE for stripping punctuation before word-search candidate fetch
- difflib.SequenceMatcher and re imported at module level

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-03-03 14:19:58 -05:00
parent 87d3f3ea16
commit eed6aedc9e
2 changed files with 116 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/services/tools.py
+105 -8
View File
@@ -2,7 +2,9 @@
import asyncio
import logging
import re
from datetime import date, datetime
from difflib import SequenceMatcher
from fabledassistant.services.caldav import (
create_event,
@@ -67,6 +69,10 @@ _CORE_TOOLS = [
"type": "string",
"description": "Optional milestone title within the project to assign this task to",
},
"confirmed": {
"type": "boolean",
"description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing task.",
},
},
"required": ["title"],
},
@@ -97,6 +103,10 @@ _CORE_TOOLS = [
"type": "string",
"description": "Optional project name to assign this note to",
},
"confirmed": {
"type": "boolean",
"description": "Set to true only when the user has explicitly confirmed creation after being warned about a similar existing note.",
},
},
"required": ["title"],
},
@@ -338,15 +348,16 @@ _CORE_TOOLS = [
"type": "function",
"function": {
"name": "create_project",
"description": "Create a new project. Use when the user asks to create a project, initiative, or campaign to organize related tasks and notes.",
"description": "Create a new project. IMPORTANT: You MUST ask the user to confirm the project name and intent before calling this tool. Only call with confirmed=true after the user has explicitly approved creation.",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Project title"},
"description": {"type": "string", "description": "Brief description"},
"goal": {"type": "string", "description": "The goal or desired outcome"},
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this project created."},
},
"required": ["title"],
"required": ["title", "confirmed"],
},
},
},
@@ -402,15 +413,16 @@ _CORE_TOOLS = [
"type": "function",
"function": {
"name": "create_milestone",
"description": "Create a new milestone inside a project for grouping and tracking related tasks.",
"description": "Create a new milestone inside a project. IMPORTANT: You MUST ask the user to confirm the milestone name and which project it belongs to before calling this tool. Only call with confirmed=true after the user has explicitly approved creation.",
"parameters": {
"type": "object",
"properties": {
"project": {"type": "string", "description": "Project title (will be created if not found)"},
"project": {"type": "string", "description": "Project title"},
"title": {"type": "string", "description": "Milestone title"},
"description": {"type": "string", "description": "Optional description"},
"confirmed": {"type": "boolean", "description": "Must be true — only set after the user has explicitly confirmed they want this milestone created."},
},
"required": ["project", "title"],
"required": ["project", "title", "confirmed"],
},
},
},
@@ -710,6 +722,27 @@ def _parse_due_date(value: str | None) -> date | None:
return None
_PUNCT_RE = re.compile(r"[^\w\s]")
def _fuzzy_title_match(title: str, candidates, threshold: float = 0.82):
"""Return (best_match, ratio) if any candidate's title is similar enough to *title*.
Uses SequenceMatcher ratio, which handles insertions/deletions/substitutions.
A threshold of 0.82 catches near-duplicates like "Game Premise" / "Game Premise Notes"
while leaving clearly different titles alone.
Returns (None, 0.0) when no candidate meets the threshold.
"""
needle = title.lower().strip()
best, best_r = None, 0.0
for c in candidates:
r = SequenceMatcher(None, needle, c.title.lower().strip()).ratio()
if r >= threshold and r > best_r:
best, best_r = c, r
return best, best_r
async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
"""Execute a tool call and return the result."""
try:
@@ -742,6 +775,27 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": False, "error": f"Milestone '{milestone_name}' not found in project '{proj.title}'. Use list_milestones to see available milestones."}
pre_fields["milestone_id"] = ms.id
existing = await get_note_by_title(user_id, task_title)
if existing is not None:
item_type = "task" if existing.status is not None else "note"
return {"success": False, "error": f"A {item_type} titled '{task_title}' already exists (id: {existing.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", task_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, limit=20)
near, ratio = _fuzzy_title_match(task_title, candidates)
if near is not None:
item_type = "task" if near.status is not None else "note"
return {"success": False, "error": f"A {item_type} with a very similar title '{near.title}' already exists (id: {near.id}, similarity: {ratio:.0%}). Use update_note to modify it instead of creating a duplicate."}
if not arguments.get("confirmed"):
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{task_title}\n{task_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.87)
if sem_hits:
best_score, best_note = sem_hits[0]
item_type = "task" if best_note.status is not None else "note"
return {"success": False, "requires_confirmation": True, "error": f"A {item_type} with very similar content exists: '{best_note.title}' (id: {best_note.id}, semantic similarity: {best_score:.0%}). Ask the user: 'An existing {item_type} covers very similar ground. Shall I still create a separate task titled \"{task_title}\"?' Then retry with confirmed=true if they agree."}
note = await create_note(
user_id=user_id,
title=task_title,
@@ -799,6 +853,27 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
project_id = proj.id
existing = await get_note_by_title(user_id, note_title)
if existing is not None:
item_type = "task" if existing.status is not None else "note"
return {"success": False, "error": f"A {item_type} titled '{note_title}' already exists (id: {existing.id}). Use update_note to modify it instead of creating a duplicate."}
clean_q = _PUNCT_RE.sub(" ", note_title).strip()
candidates, _ = await list_notes(user_id=user_id, q=clean_q, limit=20)
near, ratio = _fuzzy_title_match(note_title, candidates)
if near is not None:
item_type = "task" if near.status is not None else "note"
return {"success": False, "error": f"A {item_type} with a very similar title '{near.title}' already exists (id: {near.id}, similarity: {ratio:.0%}). Use update_note to modify it instead of creating a duplicate."}
if not arguments.get("confirmed"):
from fabledassistant.services.embeddings import semantic_search_notes as _ssn
sem_query = f"{note_title}\n{note_body}".strip()
sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.87)
if sem_hits:
best_score, best_note = sem_hits[0]
item_type = "task" if best_note.status is not None else "note"
return {"success": False, "requires_confirmation": True, "error": f"A {item_type} with very similar content exists: '{best_note.title}' (id: {best_note.id}, semantic similarity: {best_score:.0%}). Ask the user: 'An existing {item_type} covers very similar ground. Shall I still create a separate note titled \"{note_title}\"?' Then retry with confirmed=true if they agree."}
note = await create_note(
user_id=user_id,
title=note_title,
@@ -1191,8 +1266,11 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
from fabledassistant.services.projects import get_project_by_title as _gpbt, list_projects as _lp
from fabledassistant.services.milestones import create_milestone as _cm, get_project_milestone_summary
project_name = arguments.get("project", "")
ms_title_pre = arguments.get("title", "Untitled Milestone")
if not project_name:
return {"success": False, "error": "project is required"}
if not arguments.get("confirmed"):
return {"success": False, "requires_confirmation": True, "error": f"Milestone creation requires explicit user confirmation. Ask the user: 'Shall I create a milestone titled \"{ms_title_pre}\" in project \"{project_name}\"?' Then retry with confirmed=true if they agree."}
proj = await _gpbt(user_id, project_name)
if proj is None:
all_p = await _lp(user_id)
@@ -1200,10 +1278,19 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
proj = matches[0] if matches else None
if proj is None:
return {"success": False, "error": f"Project '{project_name}' not found. Use list_projects to see available projects or create_project to create one."}
ms_title = ms_title_pre
from fabledassistant.services.milestones import get_milestone_by_title as _gmbt2, list_milestones as _lms
existing_ms = await _gmbt2(user_id, proj.id, ms_title)
if existing_ms is not None:
return {"success": False, "error": f"A milestone titled '{existing_ms.title}' already exists in '{proj.title}' (id: {existing_ms.id}). Use update_milestone to modify it instead."}
all_ms = await _lms(user_id, proj.id)
near_ms, ratio = _fuzzy_title_match(ms_title, all_ms)
if near_ms is not None:
return {"success": False, "error": f"A milestone with a very similar title '{near_ms.title}' already exists in '{proj.title}' (id: {near_ms.id}, similarity: {ratio:.0%}). Use update_milestone to modify it instead."}
ms = await _cm(
user_id,
proj.id,
title=arguments.get("title", "Untitled Milestone"),
title=ms_title,
description=arguments.get("description"),
)
return {"success": True, "type": "milestone", "data": ms.to_dict()}
@@ -1253,10 +1340,20 @@ async def execute_tool(user_id: int, tool_name: str, arguments: dict) -> dict:
}
elif tool_name == "create_project":
from fabledassistant.services.projects import create_project as _create_project
from fabledassistant.services.projects import create_project as _create_project, get_project_by_title as _gpbt, list_projects as _lp
proj_title = arguments.get("title", "New Project")
if not arguments.get("confirmed"):
return {"success": False, "requires_confirmation": True, "error": f"Project creation requires explicit user confirmation. Ask the user: 'Shall I create a new project titled \"{proj_title}\"?' Then retry with confirmed=true if they agree."}
existing_proj = await _gpbt(user_id, proj_title)
if existing_proj is not None:
return {"success": False, "error": f"A project titled '{existing_proj.title}' already exists (id: {existing_proj.id}). Use update_project to modify it instead of creating a duplicate."}
all_projects = await _lp(user_id)
near_proj, ratio = _fuzzy_title_match(proj_title, all_projects)
if near_proj is not None:
return {"success": False, "error": f"A project with a very similar title '{near_proj.title}' already exists (id: {near_proj.id}, similarity: {ratio:.0%}). Use update_project to modify it instead of creating a duplicate."}
project = await _create_project(
user_id,
title=arguments.get("title", "New Project"),
title=proj_title,
description=arguments.get("description", ""),
goal=arguments.get("goal", ""),
)