moha_mind/agent/core.py
Rama master · réplica de lectura del repositorio original.
"""MohaMind Agent Core - The conversation engine.
Handles the main LLM conversation loop with tool calling.
Supports z.ai (Zhipu GLM), OpenAI, and automatic fallback.
All providers use the OpenAI-compatible API format.
"""
import asyncio
import inspect
import json
import re
from typing import Any, Awaitable, Callable, Literal, get_args, get_origin
from openai import AsyncOpenAI
from moha_mind.agent.connected_memory import ConnectedMemory
from moha_mind.agent.embeddings import build_embedder
from moha_mind.agent.energy_tracker import EnergyTracker
from moha_mind.agent.memory import MEMORY_FILES as MEMORY_FILES_KEYS
from moha_mind.agent.memory import MemoryManager
from moha_mind.agent.memory_router import MemoryRouter
from moha_mind.agent.memory_summarizer import MemorySummarizer
from moha_mind.agent.semantic_index import SemanticIndex
from moha_mind.agent.session_store import SessionStore
from moha_mind.agent.system_prompt import build_system_prompt
from moha_mind.agent.verifier import Verifier, VerifierConfig, VerifierVerdict
from moha_mind.config import settings
from moha_mind.utils.arabic_support import (
build_arabic_understanding_context,
detect_language,
normalize_colloquial_arabic,
)
from moha_mind.utils.async_work import start_daemon_worker
from moha_mind.utils.i18n import t
from moha_mind.utils.logging_config import log
from moha_mind.utils.timezone import ksa_today_str
class MohaMindAgent:
def __init__(self, memory: MemoryManager):
self.memory = memory
self.session_store = SessionStore(memory.memory_path)
self.connected_memory = ConnectedMemory(memory)
self.energy_tracker = EnergyTracker(memory)
llm_config = settings.active_llm_config
self.client = AsyncOpenAI(
api_key=llm_config["api_key"],
base_url=llm_config.get("base_url"),
)
self.model = llm_config["model"]
self.provider = llm_config.get("provider", "unknown")
self.conversations: dict[str, list[dict]] = {}
self._tool_handlers: dict[str, Callable[..., Awaitable[str]]] = {}
self._tool_schemas: dict[str, dict] = {}
# Set by bootstrap when external MCP servers are configured.
self.external_mcp: Any = None
self.strategy: str = getattr(settings, "effective_strategy", "fallback")
self.verifier: Verifier | None = self._build_verifier()
self.router = MemoryRouter(self.memory, llm_client=self.client, llm_model=self.model)
self.summarizer = MemorySummarizer(self.memory, llm_client=self.client, llm_model=self.model)
self.embedder = build_embedder()
self.semantic_index = SemanticIndex(self.memory, self.embedder)
self._memory_search_lock = asyncio.Lock()
self._memory_search_future: asyncio.Future[str] | None = None
if self.embedder:
log.info(f"Semantic memory enabled: {self.embedder.provider} ({self.embedder.model})")
log.info(
f"Agent initialized with {self.provider} (model: {self.model}) | strategy={self.strategy}"
+ (f" | verifier={self.verifier.provider}:{self.verifier.model}" if self.verifier else "")
)
def _build_verifier(self) -> Verifier | None:
if getattr(settings, "effective_strategy", "fallback") != "verify":
return None
verifier_cfg = getattr(settings, "verifier_llm_config", None)
if not verifier_cfg:
log.warning("LLM strategy is 'verify' but no verifier LLM is configured. Verification disabled.")
return None
try:
return Verifier(
VerifierConfig(
api_key=verifier_cfg["api_key"],
model=verifier_cfg["model"],
base_url=verifier_cfg.get("base_url"),
provider=verifier_cfg.get("provider", "unknown"),
strictness=getattr(settings, "verifier_strictness", "balanced"),
max_retries=max(1, int(getattr(settings, "verifier_max_retries", 1))),
)
)
except Exception as exc:
log.warning(f"Failed to build verifier: {exc}")
return None
def register_tool(self, name: str, handler: Callable[..., Awaitable[str]], schema: dict | None = None) -> None:
"""Register an external tool handler (from MCP servers).
`schema` is an optional OpenAI function-calling schema. External MCP
tools pass theirs through verbatim; without one the schema is inferred
from the handler signature.
"""
self._tool_handlers[name] = handler
if schema is not None:
self._tool_schemas[name] = schema
log.info(f"Tool registered: {name}")
def _annotation_to_schema(self, annotation: Any) -> dict:
if annotation is inspect.Signature.empty:
return {"type": "string"}
origin = get_origin(annotation)
args = get_args(annotation)
if origin is Literal:
return {"type": "string", "enum": list(args)}
if origin in (list, set, tuple):
item_schema = self._annotation_to_schema(args[0]) if args else {"type": "string"}
return {"type": "array", "items": item_schema}
if origin is dict:
return {"type": "object"}
if args and type(None) in args:
non_null = [arg for arg in args if arg is not type(None)]
if non_null:
return self._annotation_to_schema(non_null[0])
mapping = {
str: {"type": "string"},
int: {"type": "integer"},
float: {"type": "number"},
bool: {"type": "boolean"},
}
return mapping.get(annotation, {"type": "string"})
def _infer_tool_description(self, name: str, handler: Callable[..., Awaitable[str]]) -> str:
doc = inspect.getdoc(handler)
if doc:
first_line = doc.splitlines()[0].strip()
if first_line:
return first_line.rstrip(".")
return name.replace("_", " ").strip().capitalize()
def _build_dynamic_tool_schema(self, name: str, handler: Callable[..., Awaitable[str]]) -> dict:
properties = {}
required = []
for param in inspect.signature(handler).parameters.values():
if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
continue
properties[param.name] = self._annotation_to_schema(param.annotation)
if param.default is inspect.Signature.empty:
required.append(param.name)
parameters = {
"type": "object",
"properties": properties,
}
if required:
parameters["required"] = required
return {
"type": "function",
"function": {
"name": name,
"description": self._infer_tool_description(name, handler),
"parameters": parameters,
},
}
def _base_tools_schema(self) -> list[dict]:
return [
{
"type": "function",
"function": {
"name": "save_memory",
"description": (
"Append new information to a memory category without deleting existing memory. "
"Categories: profile, family, tasks, occasions, vehicle, finances, "
"health, home, documents, travel, learning, shopping, relationships, energy_log"
),
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Memory category to save to",
},
"content": {
"type": "string",
"description": "Content to save",
},
},
"required": ["category", "content"],
},
},
},
{
"type": "function",
"function": {
"name": "search_memory",
"description": "Search across all memory files for information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query",
},
"categories": {
"type": "array",
"items": {"type": "string"},
"description": "Optional: specific categories to search",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "semantic_search_memory",
"description": (
"Semantic search across memory (finds conceptually related lines, not just exact words). "
"Use when the user asks vague or paraphrased questions."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Natural-language search query"},
"top_k": {"type": "integer", "description": "Max results (default 5)"},
"categories": {
"type": "array",
"items": {"type": "string"},
"description": "Optional: restrict to these categories",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "search_sessions",
"description": "Search past conversation history for relevant details",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query for past conversations",
},
"chat_id": {
"type": "string",
"description": "Optional: limit the search to a specific chat or platform thread",
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "add_task",
"description": "Add a new task to the task list",
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Task description",
},
"priority": {
"type": "string",
"enum": ["high", "medium", "low"],
"description": "Task priority",
},
"due": {
"type": "string",
"description": "Due date in YYYY-MM-DD format",
},
},
"required": ["text"],
},
},
},
{
"type": "function",
"function": {
"name": "complete_task",
"description": "Mark a task as completed",
"parameters": {
"type": "object",
"properties": {
"task_text": {
"type": "string",
"description": "Description of the task to complete",
},
},
"required": ["task_text"],
},
},
},
{
"type": "function",
"function": {
"name": "list_tasks",
"description": "List all active (incomplete) tasks",
"parameters": {
"type": "object",
"properties": {},
},
},
},
{
"type": "function",
"function": {
"name": "get_expiring",
"description": (
"Get items expiring within a given number of days (documents, insurance, subscriptions, etc.)"
),
"parameters": {
"type": "object",
"properties": {
"days": {
"type": "integer",
"description": "Number of days to look ahead (default 90)",
},
},
},
},
},
{
"type": "function",
"function": {
"name": "get_calendar_events",
"description": "Get calendar events from Google Calendar",
"parameters": {
"type": "object",
"properties": {
"days_ahead": {
"type": "integer",
"description": "Number of days ahead to check (default 1 for today)",
},
},
},
},
},
{
"type": "function",
"function": {
"name": "append_to_section",
"description": "Append a line to a specific section in a memory file",
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Memory category",
},
"section": {
"type": "string",
"description": "Section header (without ##)",
},
"line": {
"type": "string",
"description": "Line to append",
},
},
"required": ["category", "section", "line"],
},
},
},
{
"type": "function",
"function": {
"name": "save_note",
"description": "Save a quick note to the notes folder",
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Note title",
},
"content": {
"type": "string",
"description": "Note content",
},
},
"required": ["title", "content"],
},
},
},
{
"type": "function",
"function": {
"name": "save_daily_log",
"description": "Save a daily log entry summarizing conversations or activities",
"parameters": {
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "Summary text to log",
},
},
"required": ["summary"],
},
},
},
]
def get_tools_schema(self) -> list[dict]:
"""Get the JSON schema for all available tools (OpenAI function calling format)."""
schemas = self._base_tools_schema()
known_names = {schema["function"]["name"] for schema in schemas}
for name, handler in self._tool_handlers.items():
if name in known_names:
continue
explicit = self._tool_schemas.get(name)
schemas.append(explicit if explicit is not None else self._build_dynamic_tool_schema(name, handler))
return schemas
async def _execute_model_tool_call(self, tool_call: Any) -> str:
tool_name = tool_call.function.name
log.info("Tool call: %s", tool_name)
try:
arguments = json.loads(tool_call.function.arguments)
except (json.JSONDecodeError, TypeError):
return "Error: tool arguments must be valid JSON. The tool was not executed."
if not isinstance(arguments, dict):
return "Error: tool arguments must be a JSON object. The tool was not executed."
try:
result = await asyncio.wait_for(
self.handle_tool_call(tool_name, arguments),
timeout=settings.agent_tool_timeout_seconds,
)
except TimeoutError:
log.warning("Tool timed out: %s", tool_name)
result = (
"Error: tool timed out; outcome is unknown. It may have completed externally. "
"Check state before retrying a change."
)
result = str(result)
limit = settings.agent_max_tool_result_chars
if len(result) > limit:
suffix = "\n[Tool result truncated; request a narrower query for more detail.]"
result = result[: limit - len(suffix)] + suffix
return result
async def _execute_tool_batch(self, tool_calls: list[Any], remaining_calls: int) -> list[dict]:
messages = []
for index, tool_call in enumerate(tool_calls):
if index >= remaining_calls:
result = "Error: tool-call limit reached. This tool was not executed."
else:
result = await self._execute_model_tool_call(tool_call)
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
return messages
async def handle_tool_call(self, tool_name: str, arguments: dict) -> str:
"""Handle a tool call from the LLM."""
if tool_name in self._tool_handlers:
try:
return await self._tool_handlers[tool_name](**arguments)
except Exception as e:
log.error(f"External tool '{tool_name}' failed: {e}")
return f"Error calling {tool_name}: {str(e)}"
try:
match tool_name:
case "save_memory":
self.memory.save_entry(arguments["category"], arguments["content"])
connections = self.connected_memory.process_new_info(arguments["content"])
extra = ""
if connections:
extra = " | Connected: " + "; ".join(connections)
return f"Saved to {arguments['category']}{extra}"
case "search_memory":
return await self._run_memory_search(
self._hybrid_search_memory,
arguments["query"],
arguments.get("categories"),
)
case "semantic_search_memory":
return await self._run_memory_search(
self._semantic_search_memory,
arguments["query"],
int(arguments.get("top_k") or settings.semantic_top_k or 5),
arguments.get("categories"),
)
case "search_sessions":
results = self.session_store.search_messages(
query=arguments["query"],
chat_id=arguments.get("chat_id"),
limit=arguments.get("limit", 5),
)
if not results:
return "No past conversation matches found"
return self._format_session_results(results)
case "add_task":
self.memory.add_task(
text=arguments["text"],
priority=arguments.get("priority", "medium"),
due=arguments.get("due"),
)
connections = self.connected_memory.process_new_info(arguments["text"])
return f"Task added: {arguments['text']}"
case "complete_task":
success = self.memory.complete_task(arguments["task_text"])
return "Task completed!" if success else "Task not found"
case "list_tasks":
tasks = self.memory.get_task_section()
active = [t for t in tasks if not t["done"]]
if not active:
return "No active tasks"
return "\n".join(
f"- [{t['priority'].upper()}] {t['text']}" + (f" (due {t['due']})" if t["due"] else "")
for t in active
)
case "get_expiring":
items = self.memory.get_expiring_items(arguments.get("days", 90))
if not items:
return "Nothing expiring soon"
return "\n".join(f"- [{i['days_left']}d] {i['detail']}" for i in items)
case "append_to_section":
self.memory.append_to_section(
arguments["category"],
arguments["section"],
arguments["line"],
)
return f"Added to {arguments['category']} > {arguments['section']}"
case "save_note":
path = self.memory.save_note(arguments["title"], arguments["content"])
return f"Note saved: {path.name}"
case "save_daily_log":
self.memory.save_daily_log(arguments["summary"])
return "Daily log updated"
case _:
return f"Unknown tool: {tool_name}"
except Exception as e:
log.error(f"Tool '{tool_name}' execution failed: {e}")
return f"Error: {str(e)}"
async def _run_memory_search(self, handler: Callable[..., str], *args: Any) -> str:
await self._memory_search_lock.acquire()
try:
worker = start_daemon_worker(handler, *args)
except BaseException:
self._memory_search_lock.release()
raise
self._memory_search_future = worker
worker.add_done_callback(self._finish_memory_search)
return await asyncio.shield(worker)
def _finish_memory_search(self, worker: asyncio.Future[str]) -> None:
self._memory_search_future = None
self._memory_search_lock.release()
if not worker.cancelled():
worker.exception()
def _semantic_search_memory(self, query: str, top_k: int, categories: list[str] | None) -> str:
if not self.semantic_index.is_available():
return "Semantic search is not enabled. (Set EMBEDDING_BACKEND to 'openai' or 'local'.)"
try:
self.semantic_index.sync()
except Exception as exc:
log.debug(f"Semantic sync during search failed: {exc}")
semantic = self.semantic_index.search(query, top_k=top_k, categories=categories)
if not semantic:
return "No semantic matches."
return "\n".join(f"[{r.category}:{r.line_number} · {r.score:.2f}] {r.chunk}" for r in semantic)
def _hybrid_search_memory(self, query: str, categories: list[str] | None) -> str:
"""FTS/substring results first, semantic results merged in below."""
lexical = self.memory.search(query, categories)
lexical_chunks = [f"[{r['category']}:{r['line_number']}] {r['context']}" for r in lexical[:10]]
semantic_chunks: list[str] = []
if self.semantic_index.is_available():
try:
self.semantic_index.sync()
semantic = self.semantic_index.search(
query,
top_k=int(getattr(settings, "semantic_top_k", 5) or 5),
categories=categories,
)
lexical_lines = {(r["category"], r["line_number"]) for r in lexical}
for r in semantic:
if (r.category, r.line_number) in lexical_lines:
continue
semantic_chunks.append(f"[{r.category}:{r.line_number} · sim={r.score:.2f}] {r.chunk}")
except Exception as exc:
log.debug(f"Semantic search branch failed: {exc}")
out: list[str] = []
if lexical_chunks:
out.append("Exact matches:")
out.extend(lexical_chunks)
if semantic_chunks:
if out:
out.append("")
out.append("Related (semantic):")
out.extend(semantic_chunks)
if not out:
return "No results found"
return "\n".join(out)
def _truncate_text(self, text: str, limit: int = 180) -> str:
clean = " ".join(text.split())
if len(clean) <= limit:
return clean
return clean[: limit - 3].rstrip() + "..."
def _format_session_results(self, results: list[dict]) -> str:
lines = []
for result in results:
when = result.get("created_at", "").replace("T", " ")
lines.append(f"[{result['chat_id']} | {result['role']} | {when}] {self._truncate_text(result['content'])}")
return "\n".join(lines)
def _route_memory_context(self, message: str) -> tuple[list[str] | None, dict[str, str] | None]:
"""Pick focus categories + summaries for the system prompt.
Returns (None, None) when the memory router is disabled — in that case
the prompt falls back to the legacy full-dump behavior.
"""
if not getattr(settings, "memory_router_enabled", True):
return None, None
max_cats = max(1, int(getattr(settings, "memory_router_max_categories", 4)))
try:
decision = self.router.pick(message, max_categories=max_cats)
focus = decision.categories
except Exception as exc:
log.debug(f"Memory router failed, falling back to defaults: {exc}")
focus = ["profile", "tasks", "reminders"]
summaries: dict[str, str] | None = None
if getattr(settings, "memory_summaries_enabled", True):
try:
summaries = self.summarizer.get_summaries(list(MEMORY_FILES_KEYS))
except Exception as exc:
log.debug(f"Summary fetch failed: {exc}")
summaries = None
return focus, summaries
def _build_recall_context(self, message: str, chat_id: str) -> str:
normalized_message, _ = normalize_colloquial_arabic(message)
recall_query = normalized_message if normalized_message else message
tokens = [token for token in re.findall(r"\w+", recall_query, flags=re.UNICODE) if len(token) >= 3]
if not tokens:
return ""
results = self.session_store.search_messages(recall_query, chat_id=chat_id, limit=3)
if not results:
return ""
lines = [
"### RELEVANT PAST CONVERSATION",
"Use this only when it materially helps answer the current request.",
]
for result in results:
when = result.get("created_at", "").replace("T", " ")
lines.append(f"- [{when}] {result['role']}: {self._truncate_text(result['content'], 140)}")
return "\n".join(lines)
def recall(self, query: str, chat_id: str = "default", limit: int = 5, language: str = "en") -> str:
memory_results = self.memory.search(query)[:limit]
session_results = self.session_store.search_messages(query, chat_id=chat_id, limit=limit)
if not memory_results and not session_results:
if language == "ar":
return f"لا توجد نتائج عن «{query}» في الذاكرة أو المحادثات السابقة."
return f"No results found for '{query}' in memory or past conversations."
lines = []
if memory_results:
lines.append("### الذاكرة المنظمة" if language == "ar" else "### Structured Memory")
for result in memory_results:
snippet = self._truncate_text(result["matched_line"], 120)
lines.append(f"- [{result['category']}:{result['line_number']}] {snippet}")
if session_results:
if lines:
lines.append("")
lines.append("### المحادثات السابقة" if language == "ar" else "### Past Conversations")
for result in session_results:
when = result.get("created_at", "").replace("T", " ")
lines.append(f"- [{when}] {result['role']}: {self._truncate_text(result['content'], 120)}")
return "\n".join(lines)
def _get_conversation(self, chat_id: str) -> list[dict]:
"""Get or create conversation history for a chat."""
if chat_id not in self.conversations:
persisted = self.session_store.load_recent_messages(chat_id, limit=20)
self.conversations[chat_id] = [{"role": item["role"], "content": item["content"]} for item in persisted]
return self.conversations[chat_id]
def _normalize_tool_call(self, tool_call: Any) -> dict | None:
if isinstance(tool_call, dict):
call_id = tool_call.get("id")
call_type = tool_call.get("type") or "function"
function = tool_call.get("function") or {}
else:
call_id = getattr(tool_call, "id", None)
call_type = getattr(tool_call, "type", "function") or "function"
function = getattr(tool_call, "function", None) or {}
if isinstance(function, dict):
name = function.get("name")
arguments = function.get("arguments") or "{}"
else:
name = getattr(function, "name", None)
arguments = getattr(function, "arguments", "{}") or "{}"
if not call_id or not name:
return None
if not isinstance(arguments, str):
arguments = json.dumps(arguments, ensure_ascii=False)
return {
"id": str(call_id),
"type": str(call_type),
"function": {
"name": str(name),
"arguments": arguments,
},
}
def _normalize_conversation_message(self, message: dict) -> dict | None:
role = message.get("role")
if role in ("user", "system"):
content = message.get("content")
if content is None:
return None
return {"role": role, "content": str(content)}
if role == "assistant":
tool_calls = [
normalized
for tool_call in (message.get("tool_calls") or [])
if (normalized := self._normalize_tool_call(tool_call)) is not None
]
content = message.get("content")
if tool_calls:
return {"role": "assistant", "content": content, "tool_calls": tool_calls}
if content is None:
return None
return {"role": "assistant", "content": str(content)}
if role == "tool":
tool_call_id = message.get("tool_call_id")
if not tool_call_id:
return None
return {
"role": "tool",
"tool_call_id": str(tool_call_id),
"content": str(message.get("content") or ""),
}
return None
def _valid_conversation_units(self, conversation: list[dict]) -> list[list[dict]]:
normalized = [
message
for raw_message in conversation
if (message := self._normalize_conversation_message(raw_message)) is not None
]
units: list[list[dict]] = []
index = 0
while index < len(normalized):
message = normalized[index]
if message["role"] == "assistant" and message.get("tool_calls"):
expected_ids = [tool_call["id"] for tool_call in message["tool_calls"]]
seen_ids: set[str] = set()
group = [message]
index += 1
while index < len(normalized) and normalized[index]["role"] == "tool":
tool_message = normalized[index]
tool_call_id = tool_message["tool_call_id"]
if tool_call_id in expected_ids and tool_call_id not in seen_ids:
group.append(tool_message)
seen_ids.add(tool_call_id)
index += 1
if len(seen_ids) == len(expected_ids):
units.append(group)
continue
if message["role"] != "tool":
units.append([message])
index += 1
return units
def _trim_conversation_for_llm(self, conversation: list[dict], max_messages: int = 20) -> list[dict]:
turns: list[list[dict]] = []
current_turn: list[dict] = []
for unit in self._valid_conversation_units(conversation):
if unit[0]["role"] == "user":
if current_turn:
turns.append(current_turn)
current_turn = [*unit]
else:
current_turn.extend(unit)
if current_turn:
turns.append(current_turn)
selected: list[list[dict]] = []
selected_count = 0
for turn in reversed(turns):
turn_size = len(turn)
if selected and selected_count + turn_size > max_messages:
break
selected.append(turn)
selected_count += turn_size
trimmed: list[dict] = []
for turn in reversed(selected):
trimmed.extend(turn)
return trimmed
def _build_llm_messages(self, system_prompt: str, conversation: list[dict], max_messages: int = 20) -> list[dict]:
return [
{"role": "system", "content": system_prompt},
*self._trim_conversation_for_llm(conversation, max_messages=max_messages),
]
def _switch_to_fallback(self) -> bool:
"""Switch to fallback LLM if available. Returns True if switched."""
fallback = settings.fallback_llm_config
if not fallback:
return False
if fallback.get("provider") == self.provider and fallback.get("model") == self.model:
return False
self.client = AsyncOpenAI(
api_key=fallback["api_key"],
base_url=fallback.get("base_url"),
)
self.model = fallback["model"]
self.provider = fallback.get("provider", "fallback")
log.info(f"Switched to fallback LLM: {self.provider} ({self.model})")
return True
async def _chat_completion_with_fallback(self, **kwargs):
request = dict(kwargs)
request["messages"] = list(kwargs["messages"])
tools_enabled = bool(request.get("tools")) and request.get("tool_choice") != "none"
for attempt in range(2):
response = await self._request_completion_with_fallback(**request)
message = response.choices[0].message
markup = re.search(r"<\s*/?\s*(?:tool_call|arg_key|arg_value)\b", message.content or "", re.IGNORECASE)
if not markup and (tools_enabled or not message.tool_calls):
return response
if attempt:
raise ValueError("Model returned tool markup instead of a valid response")
log.warning("Model returned invalid tool output; requesting a corrected response")
request["messages"].append(
{
"role": "user",
"content": (
"Your response used an invalid tool-call format. Never output tool_call, arg_key, "
"or arg_value XML tags. "
+ (
"Use only the provided structured function tools if you need more information; "
"otherwise provide the finished answer in the requested language."
if tools_enabled
else "No more tools are available. Write the finished answer in the requested language "
"using only the information already available. Clearly state any missing information."
)
),
}
)
async def _request_completion_with_fallback(self, **kwargs):
try:
return await self.client.chat.completions.create(model=self.model, **kwargs)
except Exception as e:
log.error(f"{self.provider} API error: {e}")
if not self._switch_to_fallback():
raise
try:
return await self.client.chat.completions.create(model=self.model, **kwargs)
except Exception as e2:
log.error(f"Fallback ({self.provider}) also failed: {e2}")
raise
async def chat(self, message: str, chat_id: str = "default") -> str:
"""Main conversation method. Process a user message and return a response."""
conversation = self._get_conversation(chat_id)
recall_context = self._build_recall_context(message, chat_id)
arabic_context = build_arabic_understanding_context(message)
extra_parts = [part for part in (arabic_context, recall_context) if part]
focus, summaries = self._route_memory_context(message)
system_prompt = build_system_prompt(
self.memory,
extra_context="\n\n".join(extra_parts),
focus_categories=focus,
summaries=summaries,
)
# Tag memory writes made during this turn as user-initiated.
self.memory.set_write_source("user")
conversation.append({"role": "user", "content": message})
self.session_store.append_message(chat_id, "user", message)
max_tool_rounds = settings.agent_max_tool_rounds
remaining_calls = settings.agent_max_tool_calls
final_response = ""
for round_num in range(max_tool_rounds):
try:
response = await self._chat_completion_with_fallback(
messages=self._build_llm_messages(system_prompt, conversation),
tools=self.get_tools_schema(),
tool_choice="auto",
max_tokens=1500,
temperature=0.7,
)
except Exception as e:
log.error(f"Chat generation failed: {e}")
return "I'm having trouble connecting right now. Please try again in a moment."
choice = response.choices[0]
assistant_message = choice.message
if assistant_message.tool_calls:
conversation.append(assistant_message.model_dump())
conversation.extend(await self._execute_tool_batch(assistant_message.tool_calls, remaining_calls))
remaining_calls -= len(assistant_message.tool_calls)
if remaining_calls <= 0:
break
else:
final_response = assistant_message.content or ""
conversation.append({"role": "assistant", "content": final_response})
break
if not final_response:
# Tool budget exhausted without a user-facing reply. Force one by
# disabling tools so the model must summarize what it already did.
log.warning("No final reply after %s rounds; forcing final reply without tools.", round_num + 1)
try:
forced_messages = self._build_llm_messages(system_prompt, conversation)
forced_messages.append(
{
"role": "user",
"content": (
"Summarize the result of my previous request for me now, in the same language "
"I used. Do not call any more tools."
),
}
)
forced = await self._chat_completion_with_fallback(
messages=forced_messages,
max_tokens=800,
temperature=0.5,
)
final_response = (forced.choices[0].message.content or "").strip()
except Exception as exc:
log.error(f"Forced final reply failed: {exc}")
if not final_response:
final_response = t("chat.done_fallback", lang=detect_language(message))
conversation.append({"role": "assistant", "content": final_response})
elif not conversation or conversation[-1].get("role") != "assistant":
conversation.append({"role": "assistant", "content": final_response})
final_response = await self._maybe_verify_and_revise(
message=message,
current_reply=final_response,
conversation=conversation,
system_prompt=system_prompt,
)
if len(conversation) > 50:
self.conversations[chat_id] = self._trim_conversation_for_llm(conversation, max_messages=30)
self.session_store.append_message(chat_id, "assistant", final_response)
self.energy_tracker.log_interaction(message, len(final_response))
return final_response
async def _maybe_verify_and_revise(
self,
*,
message: str,
current_reply: str,
conversation: list[dict],
system_prompt: str,
) -> str:
"""If the verifier strategy is enabled, ask it to check the reply and optionally retry."""
if not self.verifier or not current_reply.strip():
return current_reply
language = detect_language(message)
max_retries = max(1, self.verifier.config.max_retries)
reply = current_reply
for attempt in range(max_retries):
try:
verdict: VerifierVerdict = await self.verifier.review(message, reply, language_hint=language)
except Exception as exc:
log.warning(f"Verifier review crashed: {exc}")
return reply
if verdict.ok or verdict.severity in ("none", "low"):
if verdict.issues:
log.info(
f"Verifier accepted reply (severity={verdict.severity}) with minor notes: {verdict.issues}"
)
return reply
log.info(
f"Verifier flagged reply (severity={verdict.severity}, attempt={attempt + 1}/{max_retries}): "
f"{verdict.issues} | suggestion: {verdict.suggestion}"
)
revise_prompt = (
"A second reviewer (the verifier) checked your previous answer and found issues.\n"
f"Issues: {'; '.join(verdict.issues) or 'unspecified'}\n"
f"Reviewer suggestion: {verdict.suggestion or 'revise the reply'}\n\n"
"Rewrite your previous answer to address these issues. Keep the same language as the user. "
"Do not apologize or mention the reviewer — just produce a better answer."
)
try:
revision_messages = self._build_llm_messages(system_prompt, conversation)
revision_messages.append({"role": "user", "content": revise_prompt})
response = await self._chat_completion_with_fallback(
messages=revision_messages,
max_tokens=1500,
temperature=0.5,
)
except Exception as exc:
log.warning(f"Verifier-driven revision failed: {exc}")
return reply
new_reply = (response.choices[0].message.content or "").strip()
if not new_reply:
return reply
# Replace the last assistant turn in the conversation with the revised one.
for idx in range(len(conversation) - 1, -1, -1):
if conversation[idx].get("role") == "assistant":
conversation[idx] = {"role": "assistant", "content": new_reply}
break
reply = new_reply
return reply
async def generate_briefing(self) -> str:
"""Generate the morning briefing with one bounded batch of tool calls."""
system_prompt = build_system_prompt(self.memory, extra_context=t("briefing.mode"))
briefing_request = t("briefing.prompt")
try:
response = await self._chat_completion_with_fallback(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": briefing_request},
],
tools=self.get_tools_schema(),
tool_choice="auto",
max_tokens=2000,
temperature=0.7,
)
choice = response.choices[0]
if choice.message.tool_calls:
conversation = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": briefing_request},
choice.message.model_dump(),
]
conversation.extend(
await self._execute_tool_batch(choice.message.tool_calls, settings.agent_max_tool_calls)
)
response = await self._chat_completion_with_fallback(
messages=conversation,
max_tokens=2000,
temperature=0.7,
)
return response.choices[0].message.content or ""
return choice.message.content or ""
except Exception as e:
log.error(f"Briefing generation failed: {e}")
return t("briefing.generation_error", error=e)
async def generate_weekly_review(self) -> str:
"""Generate the weekly life review."""
system_prompt = build_system_prompt(self.memory, extra_context=t("review.mode"))
review_request = t("review.prompt")
overdue = [
task
for task in self.memory.get_task_section()
if not task["done"] and task["due"] and task["due"] < ksa_today_str()
]
cleanup = ""
if overdue:
cleanup = (
"\n\n"
+ t("review.cleanup")
+ "\n"
+ "\n".join(f"- {task['text'].strip()} ({task['due']})" for task in overdue)
)
read_tools = {
"search_memory",
"list_tasks",
"get_expiring",
"get_calendar_events",
"get_ms_calendar_events",
"social_get_neglected",
"social_get_upcoming_birthdays",
"family_get_upcoming",
}
review_tools = [tool for tool in self.get_tools_schema() if tool["function"]["name"] in read_tools]
try:
response = await self._chat_completion_with_fallback(
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": review_request},
],
tools=review_tools,
tool_choice="auto",
max_tokens=2500,
temperature=0.7,
)
choice = response.choices[0]
if choice.message.tool_calls:
conversation = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": review_request},
choice.message.model_dump(),
]
results = await self._execute_tool_batch(
[call for call in choice.message.tool_calls if call.function.name in read_tools],
settings.agent_max_tool_calls,
)
results_by_id = {result["tool_call_id"]: result for result in results}
conversation.extend(
results_by_id.get(
call.id,
{
"role": "tool",
"tool_call_id": call.id,
"content": "Error: tool not executed. Weekly reviews cannot modify saved data.",
},
)
for call in choice.message.tool_calls
)
response = await self._chat_completion_with_fallback(
messages=conversation,
max_tokens=2500,
temperature=0.7,
)
return (response.choices[0].message.content or "") + cleanup
return (choice.message.content or "") + cleanup
except Exception as e:
log.error(f"Weekly review generation failed: {e}")
return t("review.error")