Spaces:
Sleeping
Sleeping
| """Agentic Tool-Calling Workflow for DeltaMind.""" | |
| import logging, re, json | |
| from typing import Dict, Any, List | |
| from app.llm_router import llm_router | |
| from app.core.database import query_documents | |
| logger = logging.getLogger("deltamind.agent") | |
| class DeltaMindAgent: | |
| def __init__(self): | |
| self.tool_pattern = re.compile(r'\[TOOL:\s*(\w+)\((.*?)\)\]') | |
| def parse_tool_call(self, text: str) -> Dict: | |
| """Extract tool calls from LLM output.""" | |
| match = self.tool_pattern.search(text) | |
| if match: | |
| tool_name = match.group(1) | |
| args_str = match.group(2) | |
| # Simple arg parsing (assumes key=value or comma-separated) | |
| args = [a.strip() for a in args_str.split(',')] if args_str else [] | |
| return {"tool": tool_name, "args": args} | |
| return None | |
| async def execute_tool(self, tool_name: str, args: List[str]) -> str: | |
| """Mock/Simulated tool execution. In production, connect to real APIs.""" | |
| logger.info(f"Executing tool: {tool_name} with args: {args}") | |
| if tool_name == "get_weather": | |
| loc = args[0] if args else "Port Harcourt" | |
| return f"Weather in {loc}: 28°C, Light Rain, Wind 12m/s. Operational Risk: MODERATE (boat access may be delayed)." | |
| elif tool_name == "check_anomaly": | |
| return "Anomaly Check: Current parameters (Oil: 500bpd, Pressure: 2000psi) are within normal operating bounds. No immediate action required." | |
| elif tool_name == "search_knowledge": | |
| query = args[0] if args else "" | |
| res = query_documents(query, n_results=2) | |
| docs = res.get("documents", [[]])[0] | |
| return "Knowledge Base:\n" + "\n\n".join(docs) if docs else "No relevant documents found." | |
| return f"Tool {tool_name} not recognized or not implemented." | |
| async def run(self, query: str, context: str = "", history: List[Dict] = None) -> Dict: | |
| """Run the agentic loop: Think -> Act (Tool) -> Observe -> Respond.""" | |
| max_iterations = 3 | |
| current_query = query | |
| for i in range(max_iterations): | |
| result = await llm_router.generate(current_query, context=context, history=history) | |
| content = result.get("content", "") | |
| tool_call = self.parse_tool_call(content) | |
| if tool_call: | |
| logger.info(f"Agent decided to use tool: {tool_call['tool']}") | |
| # Remove the tool call from the visible content for the user | |
| clean_content = self.tool_pattern.sub('', content).strip() | |
| observation = await self.execute_tool(tool_call["tool"], tool_call["args"]) | |
| # Feed observation back to the LLM | |
| current_query = f"Previous thought: {clean_content}\nTool Observation: {observation}\nNow, provide the final answer to the user's original query: '{query}'" | |
| context = "" # Clear context to avoid token bloat | |
| continue | |
| else: | |
| # No tool call, final answer reached | |
| result["content"] = content.strip() | |
| return result | |
| return result | |
| agent = DeltaMindAgent() | |