Spaces:
Running on Zero
Running on Zero
| """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 "<think>" in content and "</think>" in content: | |
| start = content.index("<think>") + 7 | |
| end = content.index("</think>") | |
| 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 </think>." | |
| ) | |
| 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 | |