bvandeusen
77339d5c58
Merge create_task into create_note (set status='todo' for tasks, omit for notes), merge delete_task into delete_note, consolidate entity tools (create/update_person → save_person, create/update_place → save_place), rename get_note → read_note with clearer descriptions, move calculate out of rag.py into utility.py, and extract shared duplicate detection into check_duplicate() helper. Updates all downstream references in generation_task.py, quick_capture.py, ToolCallCard.vue, and WorkspaceView.vue. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""Web search, research, and image tools (require SearXNG)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from fabledassistant.services.tools._registry import tool
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@tool(
|
||
name="search_web",
|
||
description=(
|
||
"Quick web lookup — returns search result snippets for factual questions, current events, "
|
||
"definitions, or version numbers. No note is saved. Use research_topic when the user wants "
|
||
"a comprehensive written report saved as a note."
|
||
),
|
||
parameters={
|
||
"query": {"type": "string", "description": "The search query"},
|
||
},
|
||
required=["query"],
|
||
requires="searxng",
|
||
)
|
||
async def search_web_tool(*, user_id, arguments, **_ctx):
|
||
from fabledassistant.services.research import _search_searxng
|
||
|
||
query = arguments.get("query", "")
|
||
results = await _search_searxng(query)
|
||
if not results:
|
||
return {"success": False, "error": f"No results found for '{query}'"}
|
||
return {
|
||
"success": True,
|
||
"type": "web_search",
|
||
"data": {"query": query, "results": results, "count": len(results)},
|
||
}
|
||
|
||
|
||
@tool(
|
||
name="research_topic",
|
||
description=(
|
||
"Deep web research — searches multiple sources, synthesizes findings, and saves the result "
|
||
"as a structured note with sections and citations. Use when the user says 'research', "
|
||
"'look into', or wants a comprehensive write-up. Takes 30–120 seconds. "
|
||
"For a quick factual answer without saving a note, use search_web."
|
||
),
|
||
parameters={
|
||
"topic": {"type": "string", "description": "The topic or question to research"},
|
||
},
|
||
required=["topic"],
|
||
requires="searxng",
|
||
)
|
||
async def research_topic_tool(*, user_id, arguments, **_ctx):
|
||
# Research is always intercepted in generation_task.py (round 0) before execute_tool.
|
||
# Reaching here indicates a code path regression.
|
||
topic = arguments.get("topic", "")
|
||
logger.error(
|
||
"research_topic reached execute_tool — should have been intercepted upstream "
|
||
"in generation_task.py. topic=%r",
|
||
topic,
|
||
)
|
||
return {"success": False, "error": "Research could not be started. Please try again."}
|
||
|
||
|
||
@tool(
|
||
name="search_images",
|
||
description="Search and display images inline. Use ONLY when the user explicitly asks to see, show, or find an image or photo. Not for factual questions — use search_web for those.",
|
||
parameters={
|
||
"query": {"type": "string", "description": "The image search query"},
|
||
},
|
||
required=["query"],
|
||
requires="searxng",
|
||
)
|
||
async def search_images_tool(*, user_id, arguments, **_ctx):
|
||
from fabledassistant.services.images import fetch_and_store_image
|
||
from fabledassistant.services.research import _search_searxng_images
|
||
|
||
query = arguments.get("query", "")
|
||
if not query:
|
||
return {"success": False, "error": "query is required"}
|
||
raw_results = await _search_searxng_images(query)
|
||
if not raw_results:
|
||
return {"success": False, "error": f"No images found for '{query}'"}
|
||
images = []
|
||
for r in raw_results[:2]:
|
||
img_url = r.get("img_src", "")
|
||
if not img_url:
|
||
continue
|
||
record = await fetch_and_store_image(
|
||
url=img_url,
|
||
title=r.get("title"),
|
||
source_domain=r.get("source_domain"),
|
||
)
|
||
if record:
|
||
title = record.title or query
|
||
page_url = r.get("page_url", "")
|
||
source_domain = r.get("source_domain", "")
|
||
images.append({
|
||
"embed": f"",
|
||
"citation": f"*Source: [{source_domain or 'image'}]({page_url})*",
|
||
"page_url": page_url,
|
||
"title": title,
|
||
})
|
||
if not images:
|
||
return {"success": False, "error": f"Could not retrieve images for '{query}'"}
|
||
return {
|
||
"success": True,
|
||
"type": "image_search",
|
||
"data": {
|
||
"query": query,
|
||
"images": images,
|
||
"instructions": (
|
||
"Embed each image in your response by writing the 'embed' field verbatim, "
|
||
"then the 'citation' field on the next line. Do not paraphrase or list URLs."
|
||
),
|
||
},
|
||
}
|