Spaces:
Running
Running
File size: 9,922 Bytes
ceb70c7 4359d28 02ce7c3 ceb70c7 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 1262be4 02ce7c3 4359d28 b3cf31b 4359d28 ceb70c7 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 bc87248 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 b3cf31b 4359d28 ceb70c7 4359d28 b3cf31b 4359d28 ceb70c7 4359d28 ceb70c7 4359d28 |
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 |
#
# SPDX-FileCopyrightText: Hadad <hadad@linuxmail.org>
# SPDX-License-Identifier: Apache-2.0
#
import time
from assets.css.reasoning import styles
from ..response.formatter import assistant_response
from ..reasoning.interface import reasoning_interfaces
from ..reasoning.tool_reasoning import tool_reasoning
from .parser import tools_parser
from .executor import tools_call
from config import (
TOOLS_TEMPERATURE,
MAXIMUM_ITERATIONS,
MAX_RETRY_LIMIT,
ITERATION_METRICS
)
def tools_setup(server, model_name, conversation_messages, tool_definitions, search_engine):
if "current_iterations" not in ITERATION_METRICS:
ITERATION_METRICS["current_iterations"] = MAXIMUM_ITERATIONS
while ITERATION_METRICS["current_iterations"] <= MAX_RETRY_LIMIT:
for iteration_index in range(ITERATION_METRICS["current_iterations"]):
try:
if ITERATION_METRICS.get("retry_count", 0) > 0:
time.sleep(ITERATION_METRICS["retry_delays"][min(ITERATION_METRICS.get("retry_count", 0), len(ITERATION_METRICS["retry_delays"]) - 1)] * ITERATION_METRICS["backoff_multiplier"])
response = server.chat.completions.create(
model=model_name,
messages=conversation_messages,
tools=tool_definitions,
tool_choice="auto",
temperature=TOOLS_TEMPERATURE
).choices[0].message
conversation_messages.append(assistant_response(response))
if not (response.tool_calls or []):
if ITERATION_METRICS.get("logs_generator", ""):
ITERATION_METRICS["logs_generator"] = styles(ITERATION_METRICS.get("logs_generator", "").replace('<br>', '\n').strip(), expanded=False)
return conversation_messages, ITERATION_METRICS.get("logs_generator", ""), True
for tool_invocation in (response.tool_calls or []):
if tools_parser(tool_invocation.function.arguments)[1]:
ITERATION_METRICS["error_patterns"][f"{tool_invocation.function.name}_extraction"] = ITERATION_METRICS["error_patterns"].get(f"{tool_invocation.function.name}_extraction", 0) + 1
for i in range(0, len(tool_reasoning(tool_invocation.function.name, None, "error", error=tools_parser(tool_invocation.function.arguments)[1])) + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(tool_reasoning(tool_invocation.function.name, None, "error", error=tools_parser(tool_invocation.function.arguments)[1]), i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
ITERATION_METRICS["logs_generator"] = styles(tool_reasoning(tool_invocation.function.name, None, "error", error=tools_parser(tool_invocation.function.arguments)[1]), expanded=True)
yield ITERATION_METRICS["logs_generator"]
conversation_messages.append({
"role": "tool",
"tool_call_id": tool_invocation.id,
"name": tool_invocation.function.name,
"content": tools_parser(tool_invocation.function.arguments)[1]
})
else:
for i in range(0, len(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "parsing")) + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "parsing"), i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
time.sleep(ITERATION_METRICS["tools_reasoning_parsing"])
for i in range(0, len(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "executing")) + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "executing"), i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
try:
conversation_messages.append({
"role": "tool",
"tool_call_id": tool_invocation.id,
"name": tool_invocation.function.name,
"content": tools_call(
search_engine,
tool_invocation.function.name,
tools_parser(tool_invocation.function.arguments)[0]
)
})
for i in range(0, len(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "completed", result=conversation_messages[-1]["content"])) + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "completed", result=conversation_messages[-1]["content"]), i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
ITERATION_METRICS["logs_generator"] = styles(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "completed", result=conversation_messages[-1]["content"]), expanded=False)
yield ITERATION_METRICS["logs_generator"]
except Exception as tool_error:
ITERATION_METRICS["error_patterns"][f"{tool_invocation.function.name}_execution"] = ITERATION_METRICS["error_patterns"].get(f"{tool_invocation.function.name}_execution", 0) + 1
for i in range(0, len(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "error", error=str(tool_error))) + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "error", error=str(tool_error)), i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
ITERATION_METRICS["logs_generator"] = styles(tool_reasoning(tool_invocation.function.name, tools_parser(tool_invocation.function.arguments)[0], "error", error=str(tool_error)), expanded=True)
yield ITERATION_METRICS["logs_generator"]
conversation_messages.append({
"role": "tool",
"tool_call_id": tool_invocation.id,
"name": tool_invocation.function.name,
"content": str(tool_error)
})
return conversation_messages, ITERATION_METRICS.get("logs_generator", ""), True
except Exception:
ITERATION_METRICS["failures"] = ITERATION_METRICS.get("failures", 0) + 1
ITERATION_METRICS["retry_count"] = ITERATION_METRICS.get("retry_count", 0) + 1
if ITERATION_METRICS["error_patterns"]:
if max(ITERATION_METRICS["error_patterns"].values()) > 3:
ITERATION_METRICS["current_iterations"] = min(ITERATION_METRICS["current_iterations"] + 2, MAX_RETRY_LIMIT)
else:
ITERATION_METRICS["current_iterations"] = min(ITERATION_METRICS["current_iterations"] + 1, MAX_RETRY_LIMIT)
else:
ITERATION_METRICS["current_iterations"] = min(ITERATION_METRICS["current_iterations"] + 1, MAX_RETRY_LIMIT)
if ITERATION_METRICS["current_iterations"] > ITERATION_METRICS.get("previous_iterations", 0):
for i in range(0, len(f"Retrying with increased iterations: {ITERATION_METRICS['current_iterations']} (attempt {ITERATION_METRICS.get('retry_count', 0) + 1})") + 1):
ITERATION_METRICS["logs_generator"] = styles(reasoning_interfaces(f"Retrying with increased iterations: {ITERATION_METRICS['current_iterations']} (attempt {ITERATION_METRICS.get('retry_count', 0) + 1})", i), expanded=True)
yield ITERATION_METRICS["logs_generator"]
ITERATION_METRICS["previous_iterations"] = ITERATION_METRICS["current_iterations"]
if ITERATION_METRICS["current_iterations"] >= MAX_RETRY_LIMIT:
ITERATION_METRICS["logs_generator"] = styles(f"Maximum retry limit reached after {ITERATION_METRICS.get('attempts', 0)} attempts with {ITERATION_METRICS.get('failures', 0)} failures", expanded=True)
yield ITERATION_METRICS["logs_generator"]
break
ITERATION_METRICS["success_rate"] = (ITERATION_METRICS.get("tool_results_count", 0) / max(ITERATION_METRICS.get("attempts", 1), 1)) * 100
if ITERATION_METRICS.get("logs_generator", ""):
ITERATION_METRICS["logs_generator"] = styles(ITERATION_METRICS.get("logs_generator", "").replace('<br>', '\n').strip(), expanded=False)
return conversation_messages, ITERATION_METRICS.get("logs_generator", ""), ITERATION_METRICS.get("tool_results_count", 0) > 0 |