in 7-reasoning/shipmentPlanning/backend/app.py [0:0]
def call_gpt4o(plan):
gpt4o_policy_prompt = gpt4o_system_prompt.replace("{policy}", plan)
messages = [
{'role': 'system', 'content': gpt4o_policy_prompt},
]
while True:
response = client.chat.completions.create(
model='gpt-4o',
messages=messages,
tools=TOOLS,
parallel_tool_calls=False
)
assistant_message = response.choices[0].message.to_dict()
messages.append(assistant_message)
append_message({'type': 'assistant', 'content': assistant_message.get('content', '')})
if (response.choices[0].message.tool_calls and
response.choices[0].message.tool_calls[0].function.name == 'instructions_complete'):
break
if not response.choices[0].message.tool_calls:
continue
for tool in response.choices[0].message.tool_calls:
tool_id = tool.id
function_name = tool.function.name
input_arguments_str = tool.function.arguments
append_message({'type': 'tool_call', 'function_name': function_name, 'arguments': input_arguments_str})
try:
input_arguments = json.loads(input_arguments_str)
except (ValueError, json.JSONDecodeError):
continue
if function_name in function_mapping:
try:
function_response = function_mapping[function_name](**input_arguments)
except Exception as e:
function_response = {'error': str(e)}
else:
function_response = {'error': f"Function '{function_name}' not implemented."}
try:
serialized_output = json.dumps(function_response)
except (TypeError, ValueError):
serialized_output = str(function_response)
messages.append({
"role": "tool",
"tool_call_id": tool_id,
"content": serialized_output
})
append_message({'type': 'tool_response', 'function_name': function_name, 'response': serialized_output})
return messages