Flocking GPT: OpenAI LLM and rule based agents

;; VoidsGPT: Modelling Bird flocking with Generative AI

;; Author: Cristian Jimenez Romero - CY Cergy Paris University - 2025
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

extensions [ py ]

globals
[
  is-stopped?          ; flag to specify if the model is stopped
  food_collected
  step_added_distance
  step_added_heading
  overall_distances
  overall_headings
  activate_llm
]

breed [birds bird]

birds-own [
  flockmates         ;; agentset of nearby turtles
  nearest-neighbor   ;; closest one of our flockmates
  neighbors-text
  myheading
  flockmates-list
  bird-id
  action-new_heading
  action-status-ok
  action-status-code
  recovered_heading
]

to setup_birds
  clear-all
  set activate_llm true
  random-seed read-from-string used_seed

  set step_added_distance 0
  set step_added_heading 0
  set overall_distances 0
  set overall_headings 0
  create-birds 50
  [
    set bird-id who
    set size 2;
    set color yellow - 2 + random 7  ;; random shades look nice
    setxy random-xcor random-ycor
    set flockmates no-turtles
    set neighbors-text "no neighbors in vision radius"
    set myheading heading
    set shape "hawk"
 ]
  reset-ticks
end 

to go_birds  ;; forever button
  let max-separate-turn-text (word "max_separate_turn_text = '" precision (max-separate-turn) 2 "'")
  py:run max-separate-turn-text
  let max-align-turn-text (word "max_align_turn_text = '" precision (max-align-turn) 2 "'")
  py:run max-align-turn-text
  let max-cohere-turn-text (word "max_cohere_turn_text = '" precision (max-cohere-turn) 2 "'")
  py:run max-cohere-turn-text
  let minimum-separation-text (word "minimum_separation_text = '" precision (minimum-separation) 2 "'")
  py:run minimum-separation-text

  let step_text ( word "step: " ticks )
  print step_text
  ask birds
  [
    ifelse bird-id < num_gpt_birds [
      set color red
      sense-world
      ifelse activate_llm [ run_llm ][ set heading recovered_heading ]

    ]
    [
       flock
    ]

  ]
  repeat 5 [ ask turtles [ fd 0.2 ] display ]
  print "end step"
  with-local-randomness [ calculate-differences ]
  tick
end 

to flock
  find-flockmates
  if any? flockmates
    [ find-nearest-neighbor
      ifelse distance nearest-neighbor < minimum-separation
        [ separate ]
        [ align
          cohere ] ]
end 

;;; SEPARATE

to separate  ;; turtle procedure
  turn-away ([heading] of nearest-neighbor) max-separate-turn
end 

;;; ALIGN

to align  ;; turtle procedure
  if bird-id <= 0 [
    print ( word "Align: " average-flockmate-heading )
  ]
  turn-towards average-flockmate-heading max-align-turn
end 

to-report average-flockmate-heading  ;; turtle procedure
  ;; We can't just average the heading variables here.
  ;; For example, the average of 1 and 359 should be 0,
  ;; not 180.  So we have to use trigonometry.
  let x-component sum [dx] of flockmates
  let y-component sum [dy] of flockmates
  ifelse x-component = 0 and y-component = 0
    [ report heading ]
    [ report atan x-component y-component ]
end 

;;; COHERE

to cohere
  turn-towards average-heading-towards-flockmates max-cohere-turn
end 

to-report average-heading-towards-flockmates  ;; turtle procedure
  ;; "towards myself" gives us the heading from the other turtle
  ;; to me, but we want the heading from me to the other turtle,
  ;; so we add 180
  let x-component mean [sin (towards myself + 180)] of flockmates
  let y-component mean [cos (towards myself + 180)] of flockmates
  ifelse x-component = 0 and y-component = 0
    [ report heading ]
    [ report atan x-component y-component ]
end 

;;; HELPER PROCEDURES

to turn-towards [new-heading max-turn]  ;; turtle procedure
  turn-at-most (subtract-headings new-heading heading) max-turn
end 

to turn-away [new-heading max-turn]  ;; turtle procedure
  turn-at-most (subtract-headings heading new-heading) max-turn
end 

;; turn right by "turn" degrees (or left if "turn" is negative),
;; but never turn more than "max-turn" degrees

to turn-at-most [turn max-turn]  ;; turtle procedure
  ifelse abs turn > max-turn
    [ ifelse turn > 0
        [ rt max-turn ]
        [ lt max-turn ] ]
    [ rt turn ]
end 

to find-flockmates  ;; turtle procedure
  set flockmates other turtles in-radius vision
end 

to find-nearest-neighbor ;; turtle procedure
  set nearest-neighbor min-one-of flockmates [distance myself]
end 

to find-flockmates-llm  ;; turtle procedure
  set flockmates other turtles in-radius vision
  set flockmates-list []
  let my-x xcor
  let my-y ycor
  if any? flockmates [
     ask flockmates [
      let relative-x (xcor - my-x)
      let relative-y (ycor - my-y)
      let flockmate-data (list heading relative-x relative-y)
      ask myself [ set flockmates-list lput flockmate-data flockmates-list ]
     ]
  ]
end 

to find-flockmates-llm2  ;; turtle procedure
  set flockmates other turtles in-radius vision
  set flockmates-list []
  let my-x xcor
  let my-y ycor

  if any? flockmates [
    ; Create a list of flockmates with distance information
    let flockmates-with-distance []
    ask flockmates [
      let distance-to-me distance myself
      set flockmates-with-distance lput (list self distance-to-me) flockmates-with-distance
    ]

    ; Sort flockmates by distance (ascending)
    let sorted-flockmates flockmates-with-distance
    let num-flockmates length sorted-flockmates

    ; Perform bubble sort to sort the list by distance without using `?`
    let swapped true
    while [swapped] [
      set swapped false
      let i 0
      while [i < (num-flockmates - 1)] [
        let current-data item i sorted-flockmates
        let next-data item (i + 1) sorted-flockmates
        let current-distance item 1 current-data
        let next-distance item 1 next-data

        if (current-distance > next-distance) [
          ; Swap the current and next elements
          set sorted-flockmates replace-item i sorted-flockmates next-data
          set sorted-flockmates replace-item (i + 1) sorted-flockmates current-data
          set swapped true
        ]
        set i i + 1
      ]
    ]

    ; Select the closest 8 flockmates (or fewer if there aren't that many)
    let closest-flockmates n-of (min list 8 num-flockmates) sorted-flockmates

    ; Update the flockmates-list with relative positions of closest flockmates
    let closest-count length closest-flockmates
    let j 0
    repeat closest-count [
      let flockmate-data item j closest-flockmates
      let flockmate-agent item 0 flockmate-data
      let relative-x ([xcor] of flockmate-agent - my-x)
      let relative-y ([ycor] of flockmate-agent - my-y)
      let flockmate-info (list [heading] of flockmate-agent relative-x relative-y)
      set flockmates-list lput flockmate-info flockmates-list
      set j j + 1
    ]
  ]
end 

to-report generate-neighbor-info
  let info-string ""
  let counter 1
  foreach flockmates-list [
    neighbor ->
    let nheading item 0 neighbor
    let nx item 1 neighbor
    let ny item 2 neighbor
    let neighbor-info (word "neighbor_" counter ": x: " precision (nx) 2 ", y: " precision (ny) 2 ", heading: " precision (nheading) 2 " deg")
    set info-string (word info-string neighbor-info "; ")
    set counter (counter + 1)
  ]
  if info-string = "" [ set info-string "no neighbors in vision radius" ]
  report info-string
end 

to sense-world

  find-flockmates-llm
  set neighbors-text generate-neighbor-info
  set myheading ( word precision (heading) 2 )
end 

to setup
  setup_birds
  set activate_llm true
  py:setup py:python
  py:run "import math"
  py:run "import sys"
  py:run "import ollama"
  py:run "import json"
  py:run "from openai import OpenAI"
  py:run "client = OpenAI(api_key='Insert you API-key here')"
  py:run "elements_list = []"
  py:run "max_separate_turn_text = 0.0"
  py:run "max_align_turn_text = 0.0"
  py:run "max_cohere_turn_text = 0.0"
  py:run "minimum_separation_text = 0.0"
  (py:run
    "def parse_response(response):"
    "    text = response #text = response['response']"
    "    print('Raw response: ', text)"
    "    text = text.lower()"
    "    text = text.strip()"
    "    text = text.replace(chr(39), chr(34))"
    "    text = text.replace('_', '-')"
    "    parse_ok = 'True'"
    "    error_code = 'None'"
    "    try:"
    "        index = text.find( chr(34) + 'new-heading' + chr(34) + ':' )"
    "        text = text[index + 14:]"
    "        index = text.find('}')"
    "        text = text[:index]"
    "        text = text.strip()"
    "        print ('pre-processed-text: *****', text, '*****')"
    "        new_heading = text"
    "        new_heading = str(new_heading)"
    "        elements_list.append(parse_ok)"
    "        elements_list.append(error_code)"
    "        elements_list.append(new_heading.lower())"
    "        print('Parsed ok: ', elements_list)"
    "    except json.JSONDecodeError as e:"
    "        error_code = str(e)"
    "        parse_ok = 'False'"
    "        elements_list.append(parse_ok)"
    "        elements_list.append(error_code)"
    "        print ('Error: ', error_code)"
    "    except Exception as e:"
    "        error_code = str(e)"
    "        parse_ok = 'False'"
    "        elements_list.append(parse_ok)"
    "        elements_list.append(error_code)"
    "        print ('Error: ', error_code)"
    "def create_prompt(bird_heading, bird_neighbors, max_separate_turn_text, max_align_turn_text, max_cohere_turn_text, minimum_separation_text):"
    "    system_text =  'You are an agent in a 2D simulation. Following the compass convention, your task is to determine your new heading based on the flocking principles of separation turn, alignment turn (average heading of neighbors), and coherence turn (average heading towards flockmates). The parameters for these principles are: maximum-separate-turn, maximum-align-turn, maximum-cohere-turn, minimum-separation-distance. The simulation provides the following information: Current heading, Neighbors in vision radius. When calculating the alignment turn, always choose the shortest path (clockwise or counterclockwise) to align with the average heading of neighbors. Provide your final new heading after applying these rules, expressed as an angle in degrees. The result should be in JSON format only, with the keys and values: ' + chr(34) + 'rationale' + chr(34) + ' (value: your explanation) and ' + chr(34) + 'new-heading' + chr(34) + ' (value: heading in degrees). '" ;
    "    prompt_text = 'These are the flocking parameters: -Maximum separate turn: ' + max_separate_turn_text + ', -Maximum align turn: ' + max_align_turn_text + ', -Maximum cohere turn: ' + max_cohere_turn_text + ', -Minimum separation: ' + minimum_separation_text + '; This is your current environment: -Current heading: ' + bird_heading + ' deg, -Neighbors in vision radius: ' + bird_neighbors"
    "    return prompt_text, system_text"
    "def process_step(file_name, step):"
    "    # Open the text file"
    "    with open(file_name, 'r') as file:"
    "        lines = file.readlines()"
    "    # Initialize variables"
    "    in_step_section = False"
    "    in_bird_section = False"
    "    actions_list = []"
    "    bird_info = {}"
    "    # Iterate through the lines"
    "    for line in lines:"
    "        # Check for the start of the step section"
    "        if line.strip() == f'step: {step}':"
    "            in_step_section = True"
    "            continue"
    "        # Check for the end of the step section"
    "        if line.strip() == 'end step':"
    "            if in_step_section:"
    "                break"
    "            else:"
    "                continue"
    "        # If we are in the correct step section, look for bird sections"
    "        if in_step_section:"
    "            if line.startswith('Start-BirdID:'):"
    "                in_bird_section = True"
    "                bird_id = line.split(':')[1].strip()"
    "                # Initialize variables for the new ant"
    "                bird_info = {"
    "                    'BirdID': bird_id,"
    "                    'heading': 0.0"
    "                }"
    "                print('Bird ID: ', bird_id)"
    "                continue"
    "            if line.startswith('End-BirdID:'):"
    "                end_bird_id = line.split(':')[1].strip()"
    "                if in_bird_section and end_bird_id == bird_info['BirdID']:"
    "                    in_bird_section = False"
    "                    # Add the bird_info to actions_list as a list of its values"
    "                    actions_list.append(["
    "                        bird_info['BirdID'],"
    "                        bird_info['action_ok'],"
    "                        bird_info['heading']"
    "                    ])"
    "                continue"
    "            # If we are in the correct bird section, check for the required texts"
    "            if in_bird_section:"
    "                if 'Parser ok' in line:"
    "                    bird_info['action_ok'] = True"
    "                if '--- action heading:' in line:"
    "                    bird_heading = line.split(':')[1].strip()"
    "                    bird_info['heading'] = bird_heading"
    "                    print('Bird heading: ', bird_heading)"
    "    # Return the actions list"
    "    return actions_list    "
   )
end 

to-report get_llm_data
   let llm_data py:runresult "elements_list"
   report llm_data
end 

to-report populate_bird_with_llm_data [ llm_data ]
  let parse_ok item 0 llm_data
  let return_ok true
  ifelse parse_ok = "True" [
    print "Parser ok"
    set action-new_heading item 2 llm_data
    set action-status-ok true
    set action-status-code 0

    set heading ( read-from-string action-new_heading )
    print ( word "--- action heading:" read-from-string action-new_heading )
  ]
  [
    print "Parser error"
    set action-status-ok false
    set action-status-code 1
    set return_ok false
  ]
  print "end parser"
  report return_ok
end 

to run_llm
  print (word "Start-BirdID: " bird-id)
  let populate_prompt (word "prepared_prompt, system_prompt = create_prompt('" myheading "', '" neighbors-text "', max_separate_turn_text, max_align_turn_text, max_cohere_turn_text, minimum_separation_text)" )
  py:run populate_prompt
  py:run "elements_list = []"
  py:run "print('User prompt: ' + prepared_prompt)"
  py:run "print('Complete prompt: ' + system_prompt + prepared_prompt)"
  py:run "response = client.chat.completions.create(model= 'gpt-4o', max_tokens=800, timeout=30, messages=[ {'role': 'system', 'content': system_prompt}, {'role': 'user', 'content': prepared_prompt}], temperature=0.0)" ;gpt-4o-2024-05-13;
  py:run "response = response.choices[0].message.content"

  py:run "parse_response(response)"
  print "--------------- llm data: ----------------"
  carefully [
    let llm_data get_llm_data
    let populate_ok populate_bird_with_llm_data llm_data
  ]
  [
    print "Error: parsing failed!"
  ]
  print (word "End-BirdID: " bird-id)
end 

to decode_action
  py:run "elements_list = []"
  py:run "test = True"
  py:run "test = str(test)"
  py:run "elements_list.append(test)"
  py:run "elements_list.append('second')"
  let result py:runresult "elements_list"
  let item1 item 0 result
  print(item1)
  if item1 = "True" [
    print "Correct!"
  ]
end 

;; New procedure to calculate distances and heading differences between each pair of birds

to calculate-differences
  let distances [] ;; temporary list to store distances for current step
  let total-distance 0 ;; variable to store the sum of distances for the current step

  let heading-differences [] ;; temporary list to store heading differences for current step
  let total-heading-difference 0 ;; variable to store the sum of heading differences for the current step

  ;; Iterate over each turtle and calculate distances and heading differences to all other turtles
  ask turtles [
    let my-id who
    let my-heading heading
    ask other turtles [
      let other-id who
      let distance-to-other distance myself
      let heading-diff heading-difference my-heading [heading] of self
      ;print heading-diff

      ;; Store distance information
      set distances lput (list ticks my-id other-id distance-to-other) distances
      set total-distance total-distance + distance-to-other

      ;; Store heading difference information
      set heading-differences lput (list ticks my-id other-id heading-diff) heading-differences
      set total-heading-difference total-heading-difference + heading-diff
    ]
  ]
end 

;; Helper function to calculate the shortest angular difference between two headings

to-report heading-difference [heading1 heading2]
  let diff (heading1 - heading2) mod 360
  if diff > 180 [
    set diff diff - 360
  ]
  report abs diff
end