Welcome to ROSMASTER M1 repository
16. Multimodal Autonomous Agent Application
Multimodal Autonomous Agent Applications
Multimodal Autonomous Agent Applications1. Concept Introduction1.1 What is an "Autonomous Agent"?1.2 Implementation Principle Overview2. Code AnalysisKey Code1. Agent Core Workflow (largemodel/utils/ai_agent.py)2. Task Planning and LLM Interaction (largemodel/utils/ai_agent.py)3. Parameter Processing and Data Flow Implementation (largemodel/utils/ai_agent.py)Code Analysis3. Practical Operations3.1 Configuring the Offline Large Model3.1.1 Configuring the LLM Platform (yahboom.yaml)3.1.2 Configuring the Model Interface (large_model_interface.yaml)
Note: Due to performance limitations, Raspberry Pi 5 2GB, 4GB, RDK X5 4GB, and Jetson Orin Nano 4GB versions cannot run offline. Please refer to the online largemodel tutorial.
1. Concept Introduction
1.1 What is an "Autonomous Agent"?
In the largemodel project, the multimodal autonomous agent is the highest level of intelligence. It no longer simply responds to a single user command, but is capable of autonomously thinking, planning, and continuously invoking multiple tools to complete a task to achieve a complex goal.
The core of this functionality is the agent_call tool or its underlying **toolchain manager (ToolChainManager). The autonomous agent is activated when the user makes a complex request that cannot be completed with a single tool call.
1.2 Implementation Principle Overview
The autonomous agent implementation in largemodel follows the mainstream ReAct (Reason + Act) paradigm. Its core idea is to mimic the human problem-solving process, cycling between "thinking" and "acting".
- Thinking (Reason): When the agent receives a complex goal, it first invokes a powerful Language Model (LLM) to "think". It asks itself: "What should I do first to achieve this goal? Which tool should I use?" The output of the LLM is not the final answer, but an action plan.
- Action (Act): Based on the LLM's thinking results, the agent executes the corresponding action—calling
ToolsManagerto run the specified tool (such asvisual_positioning). - Observe (Observe): The agent obtains the result of the previous action ("observation"), for example,
{"result": "The cup was found, located at [120, 300, 180, 360]"}. 4. Rethinking: The agent submits the observations, along with the initial goal, to the LLM for a second round of "thinking." It asks itself, "I've found the location of the cup; what should I do next to find out its color?" The LLM might generate a new action plan, such as{"thought": "I need to analyze the image of the area where the cup is located to determine its color", "action": "seewhat", "args": {"crop_area": [120, 300, 180, 360]}}.
This thinking -> action -> observation cycle continues until the initial goal is achieved, at which point the agent generates and outputs the final answer.
2. Code Analysis
Key Code
1. Agent Core Workflow (largemodel/utils/ai_agent.py)
The _execute_agent_workflow function is the Agent's main execution loop, defining the core "planning -> execution" process.
x
# From largemodel/utils/ai_agent.pyclass AIAgent: # ... def _execute_agent_workflow(self, task_description: str) -> Dict[str, Any]: """ Executes the agent workflow: Plan -> Execute. """ try: # Step 1: Mission Planning self.node.get_logger().info("AI Agent starting task planning phase") plan_result = self._plan_task(task_description) # ... (Return early if planning fails) self.task_steps = plan_result["steps"] # Step 2: Follow all steps in order execution_results = [] tool_outputs = [] for i, step in enumerate(self.task_steps): # 2.1. Process data references in parameters before execution processed_parameters = self._process_step_parameters(step.get("parameters", {}), tool_outputs) step["parameters"] = processed_parameters # 2.2. Execute a single step step_result = self._execute_step(step, tool_outputs) execution_results.append(step_result) # 2.3. If the step succeeds, save its output for reference in subsequent steps if step_result.get("success") and step_result.get("tool_output"): tool_outputs.append(step_result["tool_output"]) else: # If any step fails, abort the entire task return { "success": False, "message": f"Task terminated because step '{step['description']}' failed." } # ... Summarize and return the final result summary = self._summarize_execution(task_description, execution_results) return { "success": True, "message": summary, "results": execution_results } # ... (Exception handling)2. Task Planning and LLM Interaction (largemodel/utils/ai_agent.py)
The core of the _plan_task function is to build a sophisticated prompt, leveraging the large model's inherent reasoning capabilities to generate a structured execution plan.
x
# From largemodel/utils/ai_agent.pyclass AIAgent: # ... def _plan_task(self, task_description: str) -> Dict[str, Any]: """ Uses the large model for task planning and decomposition. """ # Dynamically generate a list of available tools and their descriptions tool_descriptions = [] for name, adapter in self.tools_manager.tool_chain_manager.tools.items(): # ... (Get tool description from adapter.input_schema) tool_descriptions.append(f"- {name}({params}): {description}") available_tools_str = "\\n".join(tool_descriptions) # Build a highly structured plan planning_prompt = f"""As a professional task planning agent, please break down user tasks into a series of specific, executable JSON steps.**# Available Tools:**{available_tools_str}**# Core Rules:**1. **Data Passing**: When a subsequent step requires the output of a previous step, it must be referenced using the `{{{{steps.N.outputs.KEY}}}}` format.- `N` is the step ID (starting at 1).- `KEY` is the specific field name in the output data of the previous step.2. **JSON Format**: The return value must be a strict JSON object.**# User Task:**{task_description}""" # Call the large model for planning messages_to_use = [{"role": "user", "content": planning_prompt}] # Note that this calls the general text inference API. result = self.node.model_client.infer_with_text("", message=messages_to_use) # ... (Parse the JSON response and return a list of steps)3. Parameter Processing and Data Flow Implementation (largemodel/utils/ai_agent.py)
The _process_step_parameters function is responsible for parsing placeholders and implementing data flow between steps.
x
# From largemodel/utils/ai_agent.pyclass AIAgent: # ... def _process_step_parameters(self, parameters: Dict[str, Any], <