Spaces:
Running
Running
File size: 3,803 Bytes
bdafe83 53709ed bdafe83 53709ed bdafe83 |
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 |
import logging
import logging
import os
from pathlib import Path
from typing import List, Union
from llama_index.readers.file.docs import PDFReader
from agentreview.agent import Player
from .backends import IntelligenceBackend
from .config import BackendConfig
from .message import Message
class AreaChair(Player):
def __init__(
self,
name: str,
role_desc: str,
env_type: str,
backend: Union[BackendConfig, IntelligenceBackend],
global_prompt: str = None,
**kwargs,
):
super().__init__(name, role_desc, backend, global_prompt, **kwargs)
self.env_type = env_type
self.role_desc = role_desc
def act(self, observation: List[Message]) -> str:
# The author just finished their rebuttals (so last speaker is Author 1).
# The AC asks each reviewer to update their reviews.
if self.env_type == "paper_review":
if len(observation) > 0 and observation[-1].agent_name.startswith("Author"):
return "Dear reviewers, please update your reviews based on the author's rebuttals."
else:
return super().act(observation)
elif self.env_type == "paper_decision":
return super().act(observation)
else:
raise ValueError(f"Unknown env_type: {self.env_type}")
class Reviewer(Player):
def __init__(
self,
name: str,
role_desc: str,
backend: Union[BackendConfig, IntelligenceBackend],
global_prompt: str = None,
**kwargs,
):
print("kwargs")
print(kwargs)
super().__init__(name, role_desc, backend, global_prompt, **kwargs)
def act(self, observation: List[Message]) -> str:
return super().act(observation)
class PaperExtractorPlayer(Player):
"""A player for solely extracting contents from a paper.
No API calls are made by this player.
"""
def __init__(
self,
name: str,
role_desc: str,
paper_id: int,
paper_decision: str,
conference: str,
backend: Union[BackendConfig, IntelligenceBackend],
global_prompt: str = None,
**kwargs,
):
super().__init__(name, role_desc, backend, global_prompt, **kwargs)
self.paper_id = paper_id
self.paper_decision = paper_decision
self.conference: str = conference
def act(self, observation: List[Message]) -> str:
"""
Take an action based on the observation (Generate a response), which can later be parsed to actual actions that affect the game dynamics.
Parameters:
observation (List[Message]): The messages that the player has observed from the environment.
Returns:
str: The action (response) of the player.
"""
logging.info(f"Loading {self.conference} paper {self.paper_id} ({self.paper_decision}) ...")
loader = PDFReader()
document_path = Path(os.path.join(self.args.data_dir, self.conference, "paper", self.paper_decision,
f"{self.paper_id}.pdf")) #
documents = loader.load_data(file=document_path)
num_words = 0
main_contents = "Contents of this paper:\n\n"
FLAG = False
for doc in documents:
text = doc.text.split(' ')
if len(text) + num_words > self.args.max_num_words:
text = text[:self.args.max_num_words - num_words]
FLAG = True
num_words += len(text)
text = " ".join(text)
main_contents += text + ' '
if FLAG:
break
return main_contents
|