Spaces:
Running
on
L4
Running
on
L4
File size: 2,364 Bytes
ef0c954 291adda ef0c954 291adda |
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 |
from neus_v.puls.llm import *
from openai import OpenAI
from neus_v.puls.prompts import *
import json
import os
import re
def clean_and_parse_json(raw_str):
start = raw_str.find('{')
end = raw_str.rfind('}') + 1
json_str = raw_str[start:end]
return json.loads(json_str)
def process_specification(specification, propositions):
new_propositions = []
for prop in propositions:
prop_cleaned = re.sub(r"^[^a-zA-Z]+|[^a-zA-Z]+$", "", prop)
prop_cleaned = re.sub(r"\s+", "_", prop_cleaned)
new_propositions.append(prop_cleaned)
for original, new in zip(propositions, new_propositions):
specification = specification.replace(original, f'"{new}"')
replacements = {
"AND": "&",
"OR": "|",
"UNTIL": "U",
"ALWAYS": "G",
"EVENTUALLY": "F",
"NOT": "!"
}
for word, symbol in replacements.items():
specification = specification.replace(word, symbol)
return new_propositions, specification
def create_prompt(prompt, modes):
full_prompt = header + stage1_intro
for i, m in enumerate(modes, start=1):
full_prompt += f"{i}. {mode_prompts[m]}\n\n"
full_prompt += stage2_intro
full_prompt += spec_gen_intro.format(n=len(modes))
full_prompt += tl_instructions
full_prompt += input_template.format(prompt)
full_prompt += expected_output_header
for i, m in enumerate(modes):
full_prompt += mode_outputs[m]
if i != len(modes) - 1:
full_prompt += "\n\n"
full_prompt += expected_output_footer
return full_prompt
def PULS(prompt, modes, openai_key=None):
if openai_key:
os.environ["OPENAI_API_KEY"] = openai_key
client = OpenAI()
llm = LLM(client)
full_prompt = create_prompt(prompt, modes)
llm_output = llm.prompt(full_prompt)
parsed = clean_and_parse_json(llm_output)
final_output = {}
for key, value in parsed.items():
if key.endswith("_spec"):
base_key = key.replace("_spec", "")
propositions = parsed.get(base_key, [])
cleaned_props, processed_spec = process_specification(value, propositions)
final_output[base_key] = cleaned_props
final_output[key] = processed_spec
else:
final_output[key] = value
return final_output
|