slipstream/discordbot
slipstream/discordbot
1
1
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

faf

Browse Source
This commit is contained in:
Slipstream 2025-04-29 10:57:30 -06:00
parent 7735a15faf
commit d96ff7f484
Signed by: slipstream
GPG Key ID: 13E498CE010AC6FD
2 changed files with 56 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
gurt/api.py
View File

@ -1033,7 +1033,7 @@ async def get_internal_ai_json_response(
model_name: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 5000,
) -> Optional[Dict[str, Any]]: # Keep return type hint simple
) -> Optional[Tuple[Optional[Dict[str, Any]], Optional[str]]]: # Return tuple: (parsed_data, raw_text)
"""
Makes a Vertex AI call expecting a specific JSON response format for internal tasks.
@ -1047,13 +1047,16 @@ async def get_internal_ai_json_response(
max_tokens: Max output tokens.
Returns:
The parsed and validated JSON dictionary if successful, None otherwise.
A tuple containing:
- The parsed and validated JSON dictionary if successful, None otherwise.
- The raw text response received from the API, or None if the call failed before getting text.
"""
if not PROJECT_ID or not LOCATION:
print(f"Error in get_internal_ai_json_response ({task_description}): GCP Project/Location not set.")
return None
return None, None # Return tuple
final_parsed_data = None
final_parsed_data: Optional[Dict[str, Any]] = None
final_response_text: Optional[str] = None
error_occurred = None
request_payload_for_logging = {} # For logging
@ -1195,26 +1198,36 @@ async def get_internal_ai_json_response(
request_desc=task_description
)
if not response_obj or not response_obj.candidates:
raise Exception("Internal API call returned no response or candidates.")
if not response_obj:
# This case might happen if call_vertex_api_with_retry itself fails critically
raise Exception("Internal API call failed to return a response object.")
if not response_obj.candidates:
# Handle cases where the response object exists but has no candidates (e.g., safety block)
print(f"Warning: Internal API call for {task_description} returned no candidates. Response: {response_obj}")
# Attempt to get text even without candidates, though it's unlikely
final_response_text = getattr(response_obj, 'text', None)
# No valid data to parse
final_parsed_data = None
# We might still have the raw text if the API provided it despite blocking
else:
# --- Parse and Validate ---
# This function always expects JSON, so directly use response_obj.text
final_response_text = response_obj.text # Store raw text
# --- Add detailed logging for raw response text ---
print(f"--- Raw response_obj.text for {task_description} ---")
print(final_response_text)
print(f"--- End Raw response_obj.text ---")
# --- End detailed logging ---
print(f"Parsing ({task_description}): Using response_obj.text for JSON.")
# --- Parse and Validate ---
# This function always expects JSON, so directly use response_obj.text
final_response_text = response_obj.text
# --- Add detailed logging for raw response text ---
print(f"--- Raw response_obj.text for {task_description} ---")
print(final_response_text)
print(f"--- End Raw response_obj.text ---")
# --- End detailed logging ---
print(f"Parsing ({task_description}): Using response_obj.text for JSON.")
final_parsed_data = parse_and_validate_json_response(
final_response_text, response_schema_dict, f"internal task ({task_description})"
)
final_parsed_data = parse_and_validate_json_response(
final_response_text, response_schema_dict, f"internal task ({task_description})"
)
if final_parsed_data is None:
print(f"Warning: Internal task '{task_description}' failed JSON validation.")
# No re-prompting for internal tasks, just return None
if final_parsed_data is None:
print(f"Warning: Internal task '{task_description}' failed JSON validation.")
# No re-prompting for internal tasks, just return None (parsed data is None)
# Keep final_response_text as it is for returning
except Exception as e:
print(f"Error in get_internal_ai_json_response ({task_description}): {type(e).__name__}: {e}")
@ -1222,12 +1235,14 @@ async def get_internal_ai_json_response(
import traceback
traceback.print_exc()
final_parsed_data = None
# final_response_text might be None or contain partial/error text depending on when exception occurred
finally:
# Log the call
try:
# Pass the simplified payload for logging
# Pass the simplified payload and the *parsed* data for logging
await log_internal_api_call(cog, task_description, request_payload_for_logging, final_parsed_data, error_occurred)
except Exception as log_e:
print(f"Error logging internal API call: {log_e}")
return final_parsed_data
# Return both parsed data and raw text
return final_parsed_data, final_response_text

26
gurt/tools.py
View File

@ -640,7 +640,8 @@ async def _check_command_safety(cog: commands.Cog, command: str) -> Dict[str, An
{"role": "system", "content": system_prompt_content},
{"role": "user", "content": f"Analyze safety of this command: ```\n{command}\n```"}
]
safety_response = await get_internal_ai_json_response(
# Update to receive tuple: (parsed_data, raw_text)
safety_response_parsed, safety_response_raw = await get_internal_ai_json_response(
cog=cog,
prompt_messages=prompt_messages,
task_description="Command Safety Check",
@ -649,16 +650,23 @@ async def _check_command_safety(cog: commands.Cog, command: str) -> Dict[str, An
temperature=0.1,
max_tokens=150
)
if safety_response and isinstance(safety_response.get("is_safe"), bool):
is_safe = safety_response["is_safe"]
reason = safety_response.get("reason", "No reason provided.")
print(f"AI Safety Check Result: is_safe={is_safe}, reason='{reason}'")
# --- Log the raw response text ---
print(f"--- Raw AI Safety Check Response Text ---\n{safety_response_raw}\n---------------------------------------")
if safety_response_parsed and isinstance(safety_response_parsed.get("is_safe"), bool):
is_safe = safety_response_parsed["is_safe"]
reason = safety_response_parsed.get("reason", "No reason provided.")
print(f"AI Safety Check Result (Parsed): is_safe={is_safe}, reason='{reason}'")
return {"safe": is_safe, "reason": reason}
else:
# Include part of the invalid response in the error for debugging
raw_response_excerpt = str(safety_response)[:200] # Get first 200 chars
error_msg = f"AI safety check failed or returned invalid format. Response: {raw_response_excerpt}"
# Include part of the raw response in the error for debugging if parsing failed
raw_response_excerpt = str(safety_response_raw)[:200] if safety_response_raw else "N/A"
error_msg = f"AI safety check failed to parse or returned invalid format. Raw Response: {raw_response_excerpt}"
print(f"AI Safety Check Error: {error_msg}")
# Also log the parsed attempt if it exists but was invalid
if safety_response_parsed:
print(f"Parsed attempt was: {safety_response_parsed}")
return {"safe": False, "reason": error_msg}
async def run_terminal_command(cog: commands.Cog, command: str) -> Dict[str, Any]:
@ -1177,7 +1185,7 @@ async def create_new_tool(cog: commands.Cog, tool_name: str, description: str, p
try:
# Dynamically execute the generated code to define the function in the current scope
# This is risky and might fail depending on imports/scope.
# exec(python_code, globals()) # Avoid exec if possible
# thon_code, globals()) # Avoid exec if possible
# A safer way might involve importlib, but that's more complex.
# For now, just update the runtime TOOL_MAPPING if possible.