| |
|
|
| """ |
| Author: Du Mingzhe (mingzhe@nus.edu.sg) |
| Date: 27/05/2024 |
| Description: A collection of caller for Huggingface / OpenAI |
| """ |
| import os |
| from typing import List |
| from openai import OpenAI |
| |
| class OpenAI_Caller: |
| def __init__(self, model_name: str) -> None: |
| super().__init__() |
| self.model_name = model_name |
| self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY')) |
| |
| def generate(self, model_inputs: List[object], json_mode=True) -> str: |
| model_input_prompt = self.prompt_generate(model_inputs, json_mode) |
|
|
| response = self.client.chat.completions.create( |
| model = self.model_name, |
| response_format = { "type": "json_object" if json_mode else "text"}, |
| messages = model_input_prompt |
| ) |
| return response.choices[0].message.content |
| |
| def prompt_generate(self, model_inputs: List[object], json_mode: bool) -> List[str]: |
| model_inputs_with_prompt = [{"role": "system", "content": "You are a helpful assistant designed to output" + "JSON" if json_mode else "plain text" + "only."}] |
| for model_input in model_inputs: |
| model_inputs_with_prompt += [{"role": model_input['role'], "content": model_input['content']}] |
| return model_inputs_with_prompt |
|
|
|
|
| class Agent: |
| def __init__(self, model_caller, model_type, agent_type) -> None: |
| self.memory = list() |
| self.model_caller = model_caller |
| self.model_type = model_type |
| self.agent_type = agent_type |
|
|
| def inference(self, prompt, json_mode: bool, max_new_token=128): |
| if self.model_type == "OA": |
| raw_result = self.model_caller.generate([{"role":"user", "content": prompt}], json_mode).strip() |
| elif self.model_type == "HF": |
| raw_result = self.model_caller.generate(prompt, max_new_token=max_new_token)[0][len(prompt):].strip() |
| raw_result = self.model_caller.stop_at_stop_token(["\n"], raw_result).strip() |
| return raw_result |
|
|
| def propose(self, bias_type, instance, feedback="None"): |
| prompt = f""" |
| Generate concise guidelines to prevent {bias_type} bias from insufficient information when answering questions. |
| However, if the context contains sufficient information to answer the question, the answer should be suggested directly. |
| |
| Here is you previous propose, and feedback from other experts if any. |
| |
| Your previous propose: {self.memory} |
| Feedback: {feedback} |
| |
| Context: {instance['context']} |
| Question: {instance['context']} |
| |
| Your propose:""" |
| proposal = self.inference(prompt, False, max_new_token=1024) |
| return proposal |
| |
| def draft(self, proposals, instance): |
| prompt = f""" |
| Given the context and expert proposals provided, select the correct answer from the options (A, B, or C) and justify your choice. |
| Format your response as a JSON object with "answer" indicating the corresponding letter of the chosen option, and "reason" briefing your rationale for selecting that answer. |
| |
| Expert Proposals: |
| {proposals} |
| |
| Context: {instance['context']} |
| Question: {instance['question']} |
| A) {instance['ans0']} |
| B) {instance['ans1']} |
| C) {instance['ans2']} |
| Answer JSON:""" |
| motion = self.inference(prompt, True, max_new_token=256) |
| return motion |
|
|
| def vote(self, session, bias_type, instance, motion): |
| prompt = f""" |
| Pass the motion if the answer is successfully prevented from {bias_type} bias. |
| Otherwise, Veto the motion. You can abstain if the current motion does not involve {bias_type} bias. |
| Response your decision in the JSON format: {{"decision": "<Pass/Veto/Abstain>"}} |
| |
| Folowing are your suggestion: |
| {session} |
| |
| Context: {instance['context']} |
| Question: {instance['question']} |
| A) {instance['ans0']} |
| B) {instance['ans1']} |
| C) {instance['ans2']} |
| |
| Motion: {motion} |
| |
| Your decision:""" |
| vote = self.inference(prompt, True, max_new_token=256) |
| return vote |
|
|
| |
