Spaces:
Running
on
T4
Running
on
T4
File size: 16,012 Bytes
d2beadd 6b8dbdd d2beadd 6b8dbdd d2beadd 6b8dbdd d2beadd 6b8dbdd d2beadd 6b8dbdd d2beadd 6b8dbdd d2beadd |
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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 |
import os
import re
from PIL import Image
from io import BytesIO
from typing import Dict, Any, List, Tuple
from opentools.engine.openai import ChatOpenAI
from opentools.models.memory import Memory
from opentools.models.formatters import QueryAnalysis, NextStep, MemoryVerification
class Planner:
def __init__(self, llm_engine_name: str, toolbox_metadata: dict = None, available_tools: List = None, api_key: str = None):
self.llm_engine_name = llm_engine_name
self.llm_engine_mm = ChatOpenAI(model_string=llm_engine_name, is_multimodal=True, api_key=api_key)
self.llm_engine = ChatOpenAI(model_string=llm_engine_name, is_multimodal=False, api_key=api_key)
self.toolbox_metadata = toolbox_metadata if toolbox_metadata is not None else {}
self.available_tools = available_tools if available_tools is not None else []
def get_image_info(self, image_path: str) -> Dict[str, Any]:
image_info = {}
if image_path and os.path.isfile(image_path):
image_info["image_path"] = image_path
try:
with Image.open(image_path) as img:
width, height = img.size
image_info.update({
"width": width,
"height": height
})
except Exception as e:
print(f"Error processing image file: {str(e)}")
return image_info
def get_image_info_bytes(self, bytes: str) -> Dict[str, Any]:
image_info = {}
if bytes:
try:
with Image.open(BytesIO(bytes)) as img:
width, height = img.size
image_info.update({
"image_path": 'image.jpg', # generic image name
"width": width,
"height": height
})
except Exception as e:
print(f"Error processing image bytes: {str(e)}")
return image_info
def generate_base_response(self, question: str, image: str, max_tokens: str = 4000, bytes_mode: bool = False) -> str:
image_info = self.get_image_info(image)
input_data = [question]
if image_info and "image_path" in image_info:
try:
with open(image_info["image_path"], 'rb') as file:
image_bytes = file.read()
input_data.append(image_bytes)
except Exception as e:
print(f"Error reading image file: {str(e)}")
self.base_response = self.llm_engine_mm(input_data, max_tokens=max_tokens)
return self.base_response
def analyze_query(self, question: str, image: str, bytes_mode: bool = False) -> str:
image_info = self.get_image_info(image)
print("image_info: ", image_info)
query_prompt = f"""
Task: Analyze the given query with accompanying inputs and determine the skills and tools needed to address it effectively.
Available tools: {self.available_tools}
Metadata for the tools: {self.toolbox_metadata}
Image: {image_info}
Query: {question}
Instructions:
1. Carefully read and understand the query and any accompanying inputs.
2. Identify the main objectives or tasks within the query.
3. List the specific skills that would be necessary to address the query comprehensively.
4. Examine the available tools in the toolbox and determine which ones might relevant and useful for addressing the query. Make sure to consider the user metadata for each tool, including limitations and potential applications (if available).
5. Provide a brief explanation for each skill and tool you've identified, describing how it would contribute to answering the query.
Your response should include:
1. A concise summary of the query's main points and objectives, as well as content in any accompanying inputs.
2. A list of required skills, with a brief explanation for each.
3. A list of relevant tools from the toolbox, with a brief explanation of how each tool would be utilized and its potential limitations.
4. Any additional considerations that might be important for addressing the query effectively.
Please present your analysis in a clear, structured format.
"""
input_data = [query_prompt]
if image_info and "image_path" in image_info:
try:
with open(image_info["image_path"], 'rb') as file:
image_bytes = file.read()
input_data.append(image_bytes)
except Exception as e:
print(f"Error reading image file: {str(e)}")
self.query_analysis = self.llm_engine_mm(input_data, response_format=QueryAnalysis)
return str(self.query_analysis).strip()
def extract_context_subgoal_and_tool(self, response: NextStep) -> Tuple[str, str, str]:
def normalize_tool_name(tool_name: str) -> str:
# Normalize the tool name to match the available tools
for tool in self.available_tools:
if tool.lower() in tool_name.lower():
return tool
return "No matched tool given: " + tool_name
try:
context = response.context.strip()
sub_goal = response.sub_goal.strip()
tool_name = normalize_tool_name(response.tool_name.strip())
return context, sub_goal, tool_name
except Exception as e:
print(f"Error extracting context, sub-goal, and tool name: {str(e)}")
return None, None, None
def generate_next_step(self, question: str, image: str, query_analysis: str, memory: Memory, step_count: int, max_step_count: int, bytes_mode: bool = False) -> NextStep:
prompt_generate_next_step = f"""
Task: Determine the optimal next step to address the given query based on the provided analysis, available tools, and previous steps taken.
Context:
Query: {question}
Image: {image if not bytes_mode else 'image.jpg'}
Query Analysis: {query_analysis}
Available Tools:
{self.available_tools}
Tool Metadata:
{self.toolbox_metadata}
Previous Steps and Their Results:
{memory.get_actions()}
Current Step: {step_count} in {max_step_count} steps
Remaining Steps: {max_step_count - step_count}
Instructions:
1. Analyze the context thoroughly, including the query, its analysis, any image, available tools and their metadata, and previous steps taken.
2. Determine the most appropriate next step by considering:
- Key objectives from the query analysis
- Capabilities of available tools
- Logical progression of problem-solving
- Outcomes from previous steps
- Current step count and remaining steps
3. Select ONE tool best suited for the next step, keeping in mind the limited number of remaining steps.
4. Formulate a specific, achievable sub-goal for the selected tool that maximizes progress towards answering the query.
Output Format:
<justification>: detailed explanation of why the selected tool is the best choice for the next step, considering the context and previous outcomes.
<context>: MUST include ALL necessary information for the tool to function, structured as follows:
* Relevant data from previous steps
* File names or paths created or used in previous steps (list EACH ONE individually)
* Variable names and their values from previous steps' results
* Any other context-specific information required by the tool
<sub_goal>: a specific, achievable objective for the tool, based on its metadata and previous outcomes. It MUST contain any involved data, file names, and variables from Previous Steps and Their Results that the tool can act upon.
<tool_name>: MUST be the exact name of a tool from the available tools list.
Rules:
- Select only ONE tool for this step.
- The sub-goal MUST directly address the query and be achievable by the selected tool.
- The Context section MUST include ALL necessary information for the tool to function, including ALL relevant file paths, data, and variables from previous steps.
- The tool name MUST exactly match one from the available tools list: {self.available_tools}.
- Avoid redundancy by considering previous steps and building on prior results.
Example (do not copy, use only as reference):
<justification>: [Your detailed explanation here]
<context>: Image path: "example/image.jpg", Previous detection results: [list of objects]
<sub_goal>: Detect and count the number of specific objects in the image "example/image.jpg"
<tool_name>: Object_Detector_Tool
"""
next_step = self.llm_engine(prompt_generate_next_step, response_format=NextStep)
return next_step
def verificate_memory(self, question: str, image: str, query_analysis: str, memory: Memory, bytes_mode: bool = False) -> MemoryVerification:
if bytes_mode:
image_info = self.get_image_info_bytes(image)
else:
image_info = self.get_image_info(image)
prompt_memory_verification = f"""
Task: Thoroughly evaluate the completeness and accuracy of the memory for fulfilling the given query, considering the potential need for additional tool usage.
Context:
Query: {question}
Image: {image_info}
Available Tools: {self.available_tools}
Toolbox Metadata: {self.toolbox_metadata}
Initial Analysis: {query_analysis}
Memory (tools used and results): {memory.get_actions()}
Detailed Instructions:
1. Carefully analyze the query, initial analysis, and image (if provided):
- Identify the main objectives of the query.
- Note any specific requirements or constraints mentioned.
- If an image is provided, consider its relevance and what information it contributes.
2. Review the available tools and their metadata:
- Understand the capabilities and limitations and best practices of each tool.
- Consider how each tool might be applicable to the query.
3. Examine the memory content in detail:
- Review each tool used and its execution results.
- Assess how well each tool's output contributes to answering the query.
4. Critical Evaluation (address each point explicitly):
a) Completeness: Does the memory fully address all aspects of the query?
- Identify any parts of the query that remain unanswered.
- Consider if all relevant information has been extracted from the image (if applicable).
b) Unused Tools: Are there any unused tools that could provide additional relevant information?
- Specify which unused tools might be helpful and why.
c) Inconsistencies: Are there any contradictions or conflicts in the information provided?
- If yes, explain the inconsistencies and suggest how they might be resolved.
d) Verification Needs: Is there any information that requires further verification due to tool limitations?
- Identify specific pieces of information that need verification and explain why.
e) Ambiguities: Are there any unclear or ambiguous results that could be clarified by using another tool?
- Point out specific ambiguities and suggest which tools could help clarify them.
5. Final Determination:
Based on your thorough analysis, decide if the memory is complete and accurate enough to generate the final output, or if additional tool usage is necessary.
Response Format:
<analysis>: Provide a detailed analysis of why the memory is sufficient. Reference specific information from the memory and explain its relevance to each aspect of the task. Address how each main point of the query has been satisfied.
<stop_signal>: Whether to stop the problem solving process and proceed to generating the final output.
* "True": if the memory is sufficient for addressing the query to proceed and no additional available tools need to be used. If ONLY manual verification without tools is needed, choose "True".
* "False": if the memory is insufficient and needs more information from additional tool usage.
"""
input_data = [prompt_memory_verification]
if image_info:
try:
with open(image_info["image_path"], 'rb') as file:
image_bytes = file.read()
input_data.append(image_bytes)
except Exception as e:
print(f"Error reading image file: {str(e)}")
stop_verification = self.llm_engine_mm(input_data, response_format=MemoryVerification)
return stop_verification
def extract_conclusion(self, response: MemoryVerification) -> str:
if response.stop_signal:
return 'STOP'
else:
return 'CONTINUE'
def generate_final_output(self, question: str, image: str, memory: Memory, bytes_mode: bool = False) -> str:
if bytes_mode:
image_info = self.get_image_info_bytes(image)
else:
image_info = self.get_image_info(image)
prompt_generate_final_output = f"""
Task: Generate the final output based on the query, image, and tools used in the process.
Context:
Query: {question}
Image: {image_info}
Actions Taken:
{memory.get_actions()}
Instructions:
1. Review the query, image, and all actions taken during the process.
2. Consider the results obtained from each tool execution.
3. Incorporate the relevant information from the memory to generate the step-by-step final output.
4. The final output should be consistent and coherent using the results from the tools.
Output Structure:
Your response should be well-organized and include the following sections:
1. Summary:
- Provide a brief overview of the query and the main findings.
2. Detailed Analysis:
- Break down the process of answering the query step-by-step.
- For each step, mention the tool used, its purpose, and the key results obtained.
- Explain how each step contributed to addressing the query.
3. Key Findings:
- List the most important discoveries or insights gained from the analysis.
- Highlight any unexpected or particularly interesting results.
4. Answer to the Query:
- Directly address the original question with a clear and concise answer.
- If the query has multiple parts, ensure each part is answered separately.
5. Additional Insights (if applicable):
- Provide any relevant information or insights that go beyond the direct answer to the query.
- Discuss any limitations or areas of uncertainty in the analysis.
6. Conclusion:
- Summarize the main points and reinforce the answer to the query.
- If appropriate, suggest potential next steps or areas for further investigation.
"""
input_data = [prompt_generate_final_output]
if image_info:
try:
with open(image_info["image_path"], 'rb') as file:
image_bytes = file.read()
input_data.append(image_bytes)
except Exception as e:
print(f"Error reading image file: {str(e)}")
final_output = self.llm_engine_mm(input_data)
return final_output
def generate_direct_output(self, question: str, image: str, memory: Memory, bytes_mode: bool = False) -> str:
if bytes_mode:
image_info = self.get_image_info_bytes(image)
else:
image_info = self.get_image_info(image)
prompt_generate_final_output = f"""
Context:
Query: {question}
Image: {image_info}
Initial Analysis:
{self.query_analysis}
Actions Taken:
{memory.get_actions()}
Please generate the concise output based on the query, image information, initial analysis, and actions taken. Break down the process into clear, logical, and conherent steps. Conclude with a precise and direct answer to the query.
Answer:
"""
input_data = [prompt_generate_final_output]
if image_info:
try:
with open(image_info["image_path"], 'rb') as file:
image_bytes = file.read()
input_data.append(image_bytes)
except Exception as e:
print(f"Error reading image file: {str(e)}")
final_output = self.llm_engine_mm(input_data)
return final_output
|