forked from cleveragents/cleveragents-core
feat: Implement multi-agent orchestration with tool invocation
This commit is contained in:
@@ -113,6 +113,7 @@ class ToolAgent(Agent):
|
||||
tool_name = parts[0]
|
||||
tool_args = {"args": parts[1:]} if len(parts) > 1 else {}
|
||||
|
||||
|
||||
# Execute the tool
|
||||
result = await self._execute_tool(tool_name, tool_args, context)
|
||||
|
||||
@@ -133,17 +134,21 @@ class ToolAgent(Agent):
|
||||
"""Execute a specific tool."""
|
||||
# Check if tool is in the allowed list first
|
||||
dict_tool_names = [t.get("name") for t in self.tools if isinstance(t, dict)]
|
||||
|
||||
if tool_name not in self.tools and tool_name not in dict_tool_names:
|
||||
logger.error(f"Tool '{tool_name}' not in allowed list for {self.name}")
|
||||
raise ExecutionError(f"Tool '{tool_name}' not in allowed tools list")
|
||||
|
||||
# Check if it's a built-in tool
|
||||
if tool_name in self.builtin_tools:
|
||||
return await self.builtin_tools[tool_name](tool_args, context)
|
||||
result = await self.builtin_tools[tool_name](tool_args, context)
|
||||
return result
|
||||
|
||||
# Execute custom tool or shell command
|
||||
if self.allow_shell:
|
||||
return await self._execute_shell_command(tool_name, tool_args)
|
||||
else:
|
||||
logger.error(f"Tool '{tool_name}' not found and shell execution disabled for {self.name}")
|
||||
raise ExecutionError(
|
||||
f"Shell execution disabled, cannot execute '{tool_name}'"
|
||||
)
|
||||
@@ -298,20 +303,24 @@ class ToolAgent(Agent):
|
||||
content = args.get("content", "")
|
||||
|
||||
if not filepath or not content:
|
||||
logger.error(f"File write requires filepath and content")
|
||||
raise ExecutionError("File write tool requires file path and content")
|
||||
|
||||
# Check unsafe mode requirement first (can be bypassed by context)
|
||||
if not (context and context.get("_unsafe_mode", False)):
|
||||
unsafe_mode = context and context.get("_unsafe_mode", False)
|
||||
|
||||
if not unsafe_mode:
|
||||
logger.error(f"File writing requires unsafe mode")
|
||||
raise ExecutionError("File writing requires unsafe mode")
|
||||
|
||||
if self.safe_mode:
|
||||
# Always block directory traversal attempts
|
||||
if ".." in filepath:
|
||||
logger.error(f"Directory traversal blocked for {filepath}")
|
||||
raise ExecutionError("Unsafe file path blocked in safe mode")
|
||||
# Block absolute paths unless in unsafe mode
|
||||
if filepath.startswith("/") and not (
|
||||
context and context.get("_unsafe_mode", False)
|
||||
):
|
||||
if filepath.startswith("/") and not unsafe_mode:
|
||||
logger.error(f"Absolute path blocked for {filepath}")
|
||||
raise ExecutionError("Unsafe file path blocked in safe mode")
|
||||
|
||||
try:
|
||||
@@ -319,6 +328,7 @@ class ToolAgent(Agent):
|
||||
f.write(content)
|
||||
return f"Successfully wrote {len(content)} characters to {filepath}"
|
||||
except Exception as e:
|
||||
logger.error(f"File write failed for {filepath}: {e}")
|
||||
raise ExecutionError(f"File write failed: {e}")
|
||||
|
||||
def get_capabilities(self) -> List[str]:
|
||||
|
||||
@@ -294,7 +294,14 @@ class ReactiveCleverAgentsApp:
|
||||
|
||||
# Set up observers for output and errors
|
||||
def on_output(msg: StreamMessage):
|
||||
print(f"\n>>> {msg.content}")
|
||||
content_str = str(msg.content)
|
||||
|
||||
if not content_str or content_str.strip() == "":
|
||||
print(f"\n[DEBUG] Empty output received\n")
|
||||
else:
|
||||
# Check for tool execution commands
|
||||
processed_content = self._process_tool_commands(content_str)
|
||||
print(f"\n{processed_content}\n")
|
||||
|
||||
def on_error(msg: StreamMessage):
|
||||
print(f"\n[ERROR] {msg.content}")
|
||||
@@ -333,11 +340,11 @@ class ReactiveCleverAgentsApp:
|
||||
# Send user input to the stream network
|
||||
metadata: Dict[str, Any] = {"context": self.config.global_context}
|
||||
metadata["_unsafe_mode"] = self.unsafe
|
||||
|
||||
|
||||
self.stream_router.send_message("__input__", user_input, metadata)
|
||||
|
||||
# Give streams time to process
|
||||
await asyncio.sleep(0.1)
|
||||
await asyncio.sleep(2.0)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nUse 'exit' to quit.")
|
||||
@@ -657,6 +664,92 @@ class ReactiveCleverAgentsApp:
|
||||
self.langgraph_bridge.create_hybrid_pipeline(config_dict)
|
||||
self.logger.debug(f"Created hybrid pipeline: {pipeline_name}")
|
||||
|
||||
def _process_tool_commands(self, content: str) -> str:
|
||||
"""
|
||||
Process tool execution commands embedded in orchestrator output.
|
||||
|
||||
Detects [TOOL_EXECUTE:tool_name] commands and executes actual tools.
|
||||
"""
|
||||
import re
|
||||
import json
|
||||
import asyncio
|
||||
|
||||
# Pattern to match tool execution commands
|
||||
pattern = r'\[TOOL_EXECUTE:(\w+)\]\s*(\{[^}]+\})\s*\[/TOOL_EXECUTE\]'
|
||||
|
||||
def execute_tool_sync(tool_name, tool_params):
|
||||
"""Synchronous wrapper for async tool execution."""
|
||||
try:
|
||||
# Get the file_manager agent
|
||||
if 'file_manager' not in self.agents:
|
||||
self.logger.error("file_manager agent not found")
|
||||
return "Error: File manager agent not available"
|
||||
|
||||
file_manager = self.agents['file_manager']
|
||||
|
||||
# Prepare tool request in the format the tool agent expects
|
||||
tool_request = json.dumps({
|
||||
"tool": tool_name,
|
||||
"args": tool_params
|
||||
})
|
||||
|
||||
# Execute the tool
|
||||
context = {"_unsafe_mode": self.unsafe}
|
||||
|
||||
# Try to get the current running loop
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
# We're already in an async context, create a task
|
||||
import concurrent.futures
|
||||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||||
future = executor.submit(
|
||||
lambda: asyncio.run(file_manager.process_message(tool_request, context))
|
||||
)
|
||||
result = future.result(timeout=30)
|
||||
except RuntimeError:
|
||||
# No running loop, we can use asyncio.run
|
||||
result = asyncio.run(file_manager.process_message(tool_request, context))
|
||||
|
||||
return f"\n✅ {result}"
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Tool execution failed: {e}")
|
||||
return f"\n❌ Error: {str(e)}"
|
||||
|
||||
# Find and process all tool commands
|
||||
matches = list(re.finditer(pattern, content, re.DOTALL))
|
||||
|
||||
if not matches:
|
||||
# No tool commands found, return original content
|
||||
return content
|
||||
|
||||
# Process each match and build result
|
||||
result_content = content
|
||||
for match in reversed(matches): # Process in reverse to maintain positions
|
||||
tool_name = match.group(1)
|
||||
tool_params_str = match.group(2)
|
||||
|
||||
try:
|
||||
tool_params = json.loads(tool_params_str)
|
||||
tool_result = execute_tool_sync(tool_name, tool_params)
|
||||
|
||||
# Replace the tool command with the result
|
||||
result_content = (
|
||||
result_content[:match.start()] +
|
||||
tool_result +
|
||||
result_content[match.end():]
|
||||
)
|
||||
except json.JSONDecodeError as e:
|
||||
self.logger.error(f"Failed to parse tool params: {e}")
|
||||
error_msg = f"\n❌ Error: Invalid tool parameters format"
|
||||
result_content = (
|
||||
result_content[:match.start()] +
|
||||
error_msg +
|
||||
result_content[match.end():]
|
||||
)
|
||||
|
||||
return result_content
|
||||
|
||||
def _print_help(self) -> None:
|
||||
"""Print help information for interactive session."""
|
||||
print("\nAvailable commands:")
|
||||
|
||||
@@ -398,7 +398,8 @@ class LangGraph:
|
||||
await self._execute_from_node(self.config.entry_point)
|
||||
|
||||
# Return final state
|
||||
return self.state_manager.get_state()
|
||||
final_state = self.state_manager.get_state()
|
||||
return final_state
|
||||
|
||||
finally:
|
||||
self.is_running = False
|
||||
|
||||
Reference in New Issue
Block a user