Spaces:
Sleeping
Sleeping
File size: 3,293 Bytes
46e770f | 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 | """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()
|