(
self, question_answer: QuestionAnswer
)
| 230 | self._web_dir = web_dir |
| 231 | |
| 232 | def process( |
| 233 | self, question_answer: QuestionAnswer |
| 234 | ) -> Generator[QuestionAnswerEvidence, None, None]: |
| 235 | for info in question_answer.evidence_info: |
| 236 | if info.source == 'EntityPages': |
| 237 | evidence_path = os.path.join(self._wikipedia_dir, info.id) |
| 238 | elif info.source == 'SearchResult': |
| 239 | evidence_path = os.path.join(self._web_dir, info.id) |
| 240 | else: |
| 241 | raise ValueError(f'Unknown evidence source: {info.source}.') |
| 242 | with gfile.GFile(evidence_path, 'rb') as f: |
| 243 | text = f.read().decode('utf-8') |
| 244 | metrics.Metrics.counter('_', 'documents').inc() |
| 245 | yield QuestionAnswerEvidence( |
| 246 | question=question_answer.question, |
| 247 | evidence=Evidence(info=info, text=text), |
| 248 | answer=question_answer.answer) |
| 249 | |
| 250 | |
| 251 | _CLS_PIECE = '<ans>' |
nothing calls this directly
no test coverage detected