"""Synapse Agent — autonomous reasoning, planning, and tool-calling engine."""
import json
import logging
from typing import Optional, AsyncIterator
from config import config
from models.router import router
from memory.manager import MemoryManager
from agent.planner import TaskPlanner
from agent.tool_executor import ToolExecutor
logger = logging.getLogger("synapse.agent")
SYSTEM_PROMPT = """You are Synapse V8, an advanced autonomous AI assistant. You are intelligent, helpful, and capable.
You have access to the following capabilities:
- Reasoning and planning
- Memory retrieval and storage
- Tool calling (search, calculate, code execution, file analysis, etc.)
- Image description and vision analysis
- Document generation
- Internet search
When responding:
1. Think step by step for complex tasks
2. Use tools when needed — call them via the exact format provided
3. Be concise but thorough
4. If a task requires multiple steps, plan them first
5. Store important information in memory when appropriate
To call a tool, respond with:
```json
{"tool": "tool_name", "args": {"param": "value"}}
```
Available tools will be provided in the tool definitions below."""
class SynapseAgent:
"""Main agent loop — handles reasoning, tool calling, and response generation."""
def __init__(self, user_id: str = "guest", conversation_id: str = None,
mode: str = "agent", personality_prompt: str = None):
self.user_id = user_id
self.conversation_id = conversation_id
self.mode = mode
self.memory = MemoryManager(user_id)
self.planner = TaskPlanner()
self.tool_executor = ToolExecutor()
self.max_iterations = config.agent.max_iterations
self.system_prompt = personality_prompt or SYSTEM_PROMPT
async def process(self, user_message: str,
conversation_history: list[dict] = None,
mode: str = None) -> dict:
active_mode = mode or self.mode
await self.memory.add_short_term("user", user_message)
messages = self._build_messages(user_message, conversation_history)
if active_mode == "reasoning":
return await self._reasoning_loop(messages, user_message)
elif active_mode == "deep_think":
return await self._deep_think(messages, user_message)
elif active_mode == "fast":
return await self._fast_response(messages)
else:
return await self._agent_loop(messages, user_message)
async def _agent_loop(self, messages: list[dict], user_message: str) -> dict:
tool_defs = self.tool_executor.get_tool_definitions()
full_response = ""
tools_used = []
reasoning_trace = []
for iteration in range(self.max_iterations):
system_msg = self._build_system_message(tool_defs)
all_messages = [system_msg] + messages
result = await router.chat(
messages=all_messages,
temperature=config.groq.temperature,
max_tokens=config.groq.max_tokens,
)
content = result.get("content", "")
reasoning = None
if "" in content and "" in content:
start = content.index("") + 7
end = content.index("")
reasoning = content[start:end].strip()
content = content[end + 8:].strip()
reasoning_trace.append(reasoning)
tool_call = self._extract_tool_call(content)
if tool_call:
tool_name = tool_call["tool"]
tool_args = tool_call.get("args", {})
tool_result = await self.tool_executor.execute(tool_name, **tool_args)
tools_used.append({"tool": tool_name, "args": tool_args, "result": str(tool_result)[:500]})
messages.append({"role": "assistant", "content": content})
messages.append({
"role": "user",
"content": f"Tool result for {tool_name}: {json.dumps(tool_result)[:2000]}",
})
continue
full_response = content
break
await self.memory.add_short_term(
"assistant", full_response,
metadata={"mode": self.mode, "tools_used": tools_used},
)
return {
"content": full_response,
"tools_used": tools_used,
"reasoning": reasoning_trace if reasoning_trace else None,
"model": result.get("model", "unknown"),
"iterations": iteration + 1,
}
async def _reasoning_loop(self, messages: list[dict], user_message: str) -> dict:
plan = self.planner.plan(user_message)
messages.append({
"role": "system",
"content": f"Task plan:\n{json.dumps(plan, indent=2)}\nExecute this plan step by step. Show your reasoning.",
})
return await self._agent_loop(messages, user_message)
async def _deep_think(self, messages: list[dict], user_message: str) -> dict:
reflection_prompt = (
"Before answering, take a deep breath and think step by step. "
"Consider multiple perspectives. Check for edge cases. "
"Then provide your final answer after ."
)
messages.insert(0, {"role": "system", "content": reflection_prompt})
result = await self._agent_loop(messages, user_message)
reflect_messages = messages + [
{"role": "assistant", "content": result["content"]},
{"role": "user", "content": "Review your answer. Is it complete and correct? Any improvements?"},
]
reflection = await router.chat(
messages=reflect_messages,
temperature=0.3,
max_tokens=2048,
)
result["reflection"] = reflection.get("content", "")
return result
async def _fast_response(self, messages: list[dict]) -> dict:
result = await router.chat(
messages=messages,
temperature=0.5,
max_tokens=1024,
)
content = result.get("content", "")
await self.memory.add_short_term("assistant", content)
return {"content": content, "model": result.get("model", "unknown")}
def _build_messages(self, user_message: str,
conversation_history: list[dict] = None) -> list[dict]:
messages = []
if conversation_history:
messages.extend(conversation_history[-20:])
messages.append({"role": "user", "content": user_message})
return messages
def _build_system_message(self, tool_defs: list[dict] = None) -> dict:
parts = [self.system_prompt]
memory_context = self._get_sync_memory_context()
if memory_context:
parts.append(f"\n\nKnown context:\n{memory_context}")
if tool_defs:
tool_descriptions = "\n".join([
f"- {t['name']}: {t['description']}" for t in tool_defs
])
parts.append(f"\n\nAvailable tools:\n{tool_descriptions}")
return {"role": "system", "content": "\n".join(parts)}
def _get_sync_memory_context(self) -> str:
import asyncio
try:
loop = asyncio.get_event_loop()
if loop.is_running():
return ""
except RuntimeError:
pass
return ""
async def get_full_context(self) -> str:
return await self.memory.build_memory_context()
def _extract_tool_call(self, content: str) -> Optional[dict]:
markers = [
'```json\n{"tool"',
'```json\n{ "tool"',
'{"tool":',
'{"tool":',
]
for marker in markers:
if marker in content:
start = content.index(marker)
if marker.startswith("```"):
start = content.index("\n", start) + 1
end = content.find("```", start)
if end == -1:
end = len(content)
try:
return json.loads(content[start:end].strip())
except json.JSONDecodeError:
pass
return None