Spaces:
Sleeping
Sleeping
File size: 4,239 Bytes
43c34cc |
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 124 125 126 127 128 129 130 131 132 133 134 |
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],
paper_pdf_path: str = None,
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
if paper_pdf_path is not None:
self.paper_pdf_path = paper_pdf_path
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.
"""
if self.paper_pdf_path is not None:
logging.info(f"Loading paper from {self.paper_pdf_path} ...")
else:
logging.info(f"Loading {self.conference} paper {self.paper_id} ({self.paper_decision}) ...")
loader = PDFReader()
if self.paper_pdf_path is not None:
document_path = Path(self.paper_pdf_path)
else:
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
print(main_contents)
return main_contents
|