File size: 3,770 Bytes
b81ac13 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# Import tools from local extra_tools.py
try:
from extra_tools import TOOL_REGISTRY as EXTRA_TOOLS
except ImportError:
EXTRA_TOOLS = {}
# Google Sheets reading tool
def read_google_sheet(url, gid=None):
"""
Reads the first worksheet of a public Google Sheet and returns its content as a table.
"""
print("Reading Google Sheet from URL:", url)
import gspread
import pandas as pd
try:
def extract_sheet_id(url, gid=None):
import re
match = re.search(r'/d/([\w-]+)', url)
return match.group(1) if match else None
sheet_id = extract_sheet_id(url)
if gid is None:
gid = "0"
csv_url = f"https://docs.google.com/spreadsheets/d/{sheet_id}/export?format=csv&gid={gid}"
df = pd.read_csv(csv_url)
df.head()
return df.to_string(index=False)
except Exception as e:
return f"Failed to read Google Sheet: {e}"
# --- Task editing and deletion tools ---
def edit_task(task_id, description=None, deadline=None, type_=None, status=None):
"""
Edit a task's fields by its unique id. Only provided fields are updated.
"""
from app_merlin_ai_coach import session_memory # Import moved inside function
for t in session_memory.tasks:
if t.get("id") == task_id:
if description is not None:
t["description"] = description
if deadline is not None:
t["deadline"] = deadline
if type_ is not None:
t["type"] = type_
if status is not None:
t["status"] = status
return f"Task {task_id} updated."
return f"Task {task_id} not found."
def delete_task(task_id):
"""
Delete a task by its unique id.
"""
from app_merlin_ai_coach import session_memory # Import moved inside function
before = len(session_memory.tasks)
session_memory.tasks = [t for t in session_memory.tasks if t.get("id") != task_id]
after = len(session_memory.tasks)
if before == after:
return f"Task {task_id} not found."
return f"Task {task_id} deleted."
TOOL_REGISTRY = {
**EXTRA_TOOLS,
"read_google_sheet": {
"description": "Read a public Google Sheet and return its content as a table. Usage: read_google_sheet(url, gid (Optional))",
"function": read_google_sheet,
},
"edit_task": {
"description": "Edit a task by id. Usage: edit_task(task_id, description=..., deadline=..., type_=..., status=...). Only provide fields you want to change.",
"function": edit_task,
},
"delete_task": {
"description": "Delete a task by id. Usage: delete_task(task_id)",
"function": delete_task,
},
# Add more tools here as needed
}
def call_tool(tool_name, *args, **kwargs):
"""
Calls a registered tool by name.
"""
tool = TOOL_REGISTRY.get(tool_name)
if not tool:
return f"Tool '{tool_name}' not found."
try:
return tool["function"](*args, **kwargs)
except Exception as e:
return f"Error running tool '{tool_name}': {e}"
def get_tool_descriptions():
"""
Returns a string describing all available tools for the system prompt.
"""
descs = []
# Add system instruction about no nested tool calls
# descs.append("System instruction: Tool calls cannot be nested. Do not call a tool/function within another tool/function call.")
for name, tool in TOOL_REGISTRY.items():
descs.append(f"{name}: {tool['description']}")
return "\n".join(descs)
def get_tool_functions():
"""
Returns a list of tool functions for use with LangChain/LangGraph ToolNode.
"""
return [tool["function"] for tool in TOOL_REGISTRY.values()]
|