Coverage for src/inheritance_calculator_core/agents/interview_agent.py: 0%

1"""相続情報収集インタビューエージェント 

2 

3AIを使って対話形式で相続情報を収集するエージェント。 

4""" 

5from typing import Dict, List, Optional, Any 

6from datetime import date, datetime 

7from enum import Enum 

8import logging 

9import re 

10 

11from .ollama_client import OllamaClient 

12from .prompts import InheritancePrompts 

13from ..models.person import Person, Gender 

14from ..models.relationship import BloodType 

15 

16 

17class InterviewState(str, Enum): 

18 """インタビューの状態""" 

19 INIT = "init" 

20 DECEDENT_INFO = "decedent_info" 

21 SPOUSE_INFO = "spouse_info" 

22 CHILDREN_INFO = "children_info" 

23 PARENTS_INFO = "parents_info" 

24 SIBLINGS_INFO = "siblings_info" 

25 SPECIAL_CASES = "special_cases" 

26 CONFIRMATION = "confirmation" 

27 COMPLETED = "completed" 

28 

29 

30class InterviewAgent: 

31 """ 

32 対話型相続情報収集エージェント 

33 

34 ユーザーと対話しながら相続に必要な情報を収集し、 

35 Personオブジェクトとして構造化する。 

36 """ 

37 

38 def __init__(self, ollama_client: Optional[OllamaClient] = None) -> None: 

39 """ 

40 初期化 

41 

42 Args: 

43 ollama_client: Ollamaクライアント（Noneの場合は新規作成） 

44 """ 

45 self.logger = logging.getLogger(__name__) 

46 self.client = ollama_client or OllamaClient() 

47 self.prompts = InheritancePrompts() 

48 

49 # 状態管理 

50 self.state = InterviewState.INIT 

51 self.collected_data: Dict[str, Any] = {} 

52 self.conversation_history: List[Dict[str, str]] = [] 

53 

54 # 収集したデータ 

55 self.decedent: Optional[Person] = None 

56 self.spouses: List[Person] = [] 

57 self.children: List[Person] = [] 

58 self.parents: List[Person] = [] 

59 self.siblings: List[Person] = [] 

60 self.renounced: List[Person] = [] 

61 self.disqualified: List[Person] = [] 

62 self.disinherited: List[Person] = [] 

63 self.sibling_blood_types: Dict[str, BloodType] = {} 

64 self.retransfer_heirs_info: Dict[str, List[Person]] = {} 

65 

66 self.logger.info("InterviewAgent initialized") 

67 

68 def start_interview(self) -> str: 

69 """ 

70 インタビューを開始 

71 

72 Returns: 

73 最初の質問メッセージ 

74 """ 

75 self.state = InterviewState.DECEDENT_INFO 

76 self.logger.info("Interview started") 

77 

78 welcome_message = self.prompts.DECEDENT_INTRO + "\n\n" + self.prompts.DECEDENT_NAME 

79 

80 # 会話履歴に追加 

81 self.conversation_history.append({ 

82 "role": "assistant", 

83 "content": welcome_message 

84 }) 

85 

86 return welcome_message 

87 

88 def process_response(self, user_input: str) -> str: 

89 """ 

90 ユーザーの応答を処理し、次の質問を返す 

91 

92 Args: 

93 user_input: ユーザーの入力 

94 

95 Returns: 

96 次の質問または確認メッセージ 

97 """ 

98 # 会話履歴に追加 

99 self.conversation_history.append({ 

100 "role": "user", 

101 "content": user_input 

102 }) 

103 

104 try: 

105 # 状態に応じて処理を分岐 

106 if self.state == InterviewState.DECEDENT_INFO: 

107 next_message = self._process_decedent_info(user_input) 

108 elif self.state == InterviewState.SPOUSE_INFO: 

109 next_message = self._process_spouse_info(user_input) 

110 elif self.state == InterviewState.CHILDREN_INFO: 

111 next_message = self._process_children_info(user_input) 

112 elif self.state == InterviewState.PARENTS_INFO: 

113 next_message = self._process_parents_info(user_input) 

114 elif self.state == InterviewState.SIBLINGS_INFO: 

115 next_message = self._process_siblings_info(user_input) 

116 elif self.state == InterviewState.SPECIAL_CASES: 

117 next_message = self._process_special_cases(user_input) 

118 elif self.state == InterviewState.CONFIRMATION: 

119 next_message = self._process_confirmation(user_input) 

120 else: 

121 next_message = "インタビューは完了しました。" 

122 

123 # 会話履歴に追加 

124 self.conversation_history.append({ 

125 "role": "assistant", 

126 "content": next_message 

127 }) 

128 

129 return next_message 

130 

131 except Exception as e: 

132 self.logger.error(f"Error processing response: {e}", exc_info=True) 

133 error_message = "申し訳ございません。エラーが発生しました。もう一度お答えいただけますか？" 

134 self.conversation_history.append({ 

135 "role": "assistant", 

136 "content": error_message 

137 }) 

138 return error_message 

139 

140 def _process_decedent_info(self, user_input: str) -> str: 

141 """被相続人情報を処理""" 

142 # まだ名前を収集していない 

143 if "decedent_name" not in self.collected_data: 

144 self.collected_data["decedent_name"] = user_input.strip() 

145 return self.prompts.DECEDENT_DEATH_DATE 

146 

147 # まだ死亡日を収集していない 

148 if "decedent_death_date" not in self.collected_data: 

149 death_date = self._parse_date(user_input) 

150 if death_date: 

151 self.collected_data["decedent_death_date"] = death_date 

152 return self.prompts.DECEDENT_BIRTH_DATE 

153 else: 

154 return "申し訳ございません。日付の形式が正しくありません。\n" + self.prompts.DECEDENT_DEATH_DATE 

155 

156 # まだ生年月日を収集していない 

157 if "decedent_birth_date" not in self.collected_data: 

158 if user_input.strip() in ["不明", "わからない", "分からない"]: 

159 birth_date = None 

160 else: 

161 birth_date = self._parse_date(user_input) 

162 

163 self.collected_data["decedent_birth_date"] = birth_date 

164 

165 # 被相続人オブジェクトを作成 

166 self.decedent = Person( 

167 name=self.collected_data["decedent_name"], 

168 is_decedent=True, 

169 is_alive=False, 

170 death_date=self.collected_data["decedent_death_date"], 

171 birth_date=birth_date 

172 ) 

173 

174 # 次の状態へ 

175 self.state = InterviewState.SPOUSE_INFO 

176 return "\n" + self.prompts.SPOUSE_QUESTION 

177 

178 return "エラーが発生しました。" 

179 

180 def _process_spouse_info(self, user_input: str) -> str: 

181 """配偶者情報を処理""" 

182 # 配偶者の有無を確認 

183 if "has_spouse" not in self.collected_data: 

184 has_spouse = self._parse_yes_no(user_input) 

185 self.collected_data["has_spouse"] = has_spouse 

186 

187 if has_spouse: 

188 return self.prompts.SPOUSE_INFO 

189 else: 

190 # 次の状態へ 

191 self.state = InterviewState.CHILDREN_INFO 

192 return "\n" + self.prompts.CHILDREN_QUESTION 

193 

194 # 配偶者情報を収集中 

195 # LLMで構造化データ抽出 

196 spouse_info = self._extract_person_info(user_input, "配偶者") 

197 spouse = Person( 

198 name=spouse_info.get("name", "配偶者"), 

199 is_alive=spouse_info.get("is_alive", True), 

200 birth_date=spouse_info.get("birth_date"), 

201 gender=spouse_info.get("gender", Gender.UNKNOWN) 

202 ) 

203 self.spouses.append(spouse) 

204 

205 # 次の状態へ 

206 self.state = InterviewState.CHILDREN_INFO 

207 return "\n" + self.prompts.CHILDREN_QUESTION 

208 

209 def _process_children_info(self, user_input: str) -> str: 

210 """子の情報を処理""" 

211 if "has_children" not in self.collected_data: 

212 has_children = self._parse_yes_no(user_input) 

213 self.collected_data["has_children"] = has_children 

214 

215 if has_children: 

216 return self.prompts.CHILDREN_COUNT 

217 else: 

218 # 次の状態へ 

219 self.state = InterviewState.PARENTS_INFO 

220 return "\n" + self.prompts.PARENTS_QUESTION 

221 

222 # 子の人数を確認 

223 if "children_count" not in self.collected_data: 

224 try: 

225 count = int(user_input.strip()) 

226 self.collected_data["children_count"] = count 

227 self.collected_data["children_collected"] = 0 

228 return self.prompts.format_child_info(1) 

229 except ValueError: 

230 return "申し訳ございません。数字でご入力ください。\n" + self.prompts.CHILDREN_COUNT 

231 

232 # 子の情報を収集中 

233 collected = self.collected_data["children_collected"] 

234 total = self.collected_data["children_count"] 

235 

236 # LLMで構造化データ抽出 

237 child_info = self._extract_person_info(user_input, f"子{collected + 1}") 

238 child = Person( 

239 name=child_info.get("name", f"子{collected + 1}"), 

240 is_alive=child_info.get("is_alive", True), 

241 birth_date=child_info.get("birth_date"), 

242 death_date=child_info.get("death_date"), 

243 gender=child_info.get("gender", Gender.UNKNOWN) 

244 ) 

245 self.children.append(child) 

246 

247 collected += 1 

248 self.collected_data["children_collected"] = collected 

249 

250 if collected < total: 

251 return self.prompts.format_child_info(collected + 1) 

252 else: 

253 # 次の状態へ 

254 self.state = InterviewState.PARENTS_INFO 

255 return "\n" + self.prompts.PARENTS_QUESTION 

256 

257 def _process_parents_info(self, user_input: str) -> str: 

258 """直系尊属情報を処理""" 

259 if "has_parents" not in self.collected_data: 

260 has_parents = self._parse_yes_no(user_input) 

261 self.collected_data["has_parents"] = has_parents 

262 

263 if not has_parents: 

264 # 次の状態へ 

265 self.state = InterviewState.SIBLINGS_INFO 

266 return "\n" + self.prompts.SIBLINGS_QUESTION 

267 

268 # 親の人数を聞く 

269 self.collected_data["parents"] = [] 

270 return "存命の直系尊属（父母、祖父母など）は何人いますか？" 

271 

272 # 親の情報を収集 

273 if "parent_count" not in self.collected_data: 

274 try: 

275 count = int(user_input.strip()) 

276 self.collected_data["parent_count"] = count 

277 self.collected_data["parent_collected"] = 0 

278 

279 if count == 0: 

280 self.state = InterviewState.SIBLINGS_INFO 

281 return "\n" + self.prompts.SIBLINGS_QUESTION 

282 

283 return f"1人目の直系尊属について教えてください。\n氏名、続柄（父、母、祖父など）、生年月日などを入力してください。" 

284 except ValueError: 

285 return "数字で入力してください。存命の直系尊属は何人いますか？" 

286 

287 # 親の詳細情報を収集 

288 collected = self.collected_data["parent_collected"] 

289 count = self.collected_data["parent_count"] 

290 

291 parent_info = self._extract_person_info(user_input, f"直系尊属{collected + 1}") 

292 parent = Person( 

293 name=parent_info.get("name", f"直系尊属{collected + 1}"), 

294 is_alive=True, # 存命の直系尊属のみ 

295 birth_date=parent_info.get("birth_date"), 

296 gender=parent_info.get("gender", Gender.UNKNOWN) 

297 ) 

298 self.collected_data["parents"].append(parent) 

299 self.collected_data["parent_collected"] = collected + 1 

300 

301 if collected + 1 < count: 

302 return f"{collected + 2}人目の直系尊属について教えてください。\n氏名、続柄、生年月日などを入力してください。" 

303 

304 # 全員収集完了 

305 self.state = InterviewState.SIBLINGS_INFO 

306 return "\n" + self.prompts.SIBLINGS_QUESTION 

307 

308 def _process_siblings_info(self, user_input: str) -> str: 

309 """兄弟姉妹情報を処理""" 

310 if "has_siblings" not in self.collected_data: 

311 has_siblings = self._parse_yes_no(user_input) 

312 self.collected_data["has_siblings"] = has_siblings 

313 

314 if not has_siblings: 

315 # 次の状態へ（特殊ケース） 

316 self.state = InterviewState.SPECIAL_CASES 

317 self.collected_data["special_case_step"] = "renunciation" 

318 return "\n" + self.prompts.RENUNCIATION_QUESTION 

319 

320 # 兄弟姉妹の人数を聞く 

321 self.collected_data["siblings"] = [] 

322 return "存命の兄弟姉妹は何人いますか？" 

323 

324 # 兄弟姉妹の情報を収集 

325 if "sibling_count" not in self.collected_data: 

326 try: 

327 count = int(user_input.strip()) 

328 self.collected_data["sibling_count"] = count 

329 self.collected_data["sibling_collected"] = 0 

330 

331 if count == 0: 

332 self.state = InterviewState.SPECIAL_CASES 

333 self.collected_data["special_case_step"] = "renunciation" 

334 return "\n" + self.prompts.RENUNCIATION_QUESTION 

335 

336 return f"1人目の兄弟姉妹について教えてください。\n氏名、続柄（兄、姉、弟、妹など）、血縁タイプ（全血・半血）、生年月日などを入力してください。" 

337 except ValueError: 

338 return "数字で入力してください。存命の兄弟姉妹は何人いますか？" 

339 

340 # 兄弟姉妹の詳細情報を収集 

341 collected = self.collected_data["sibling_collected"] 

342 count = self.collected_data["sibling_count"] 

343 

344 sibling_info = self._extract_person_info(user_input, f"兄弟姉妹{collected + 1}") 

345 sibling = Person( 

346 name=sibling_info.get("name", f"兄弟姉妹{collected + 1}"), 

347 is_alive=True, # 存命の兄弟姉妹のみ 

348 birth_date=sibling_info.get("birth_date"), 

349 gender=sibling_info.get("gender", Gender.UNKNOWN) 

350 ) 

351 self.collected_data["siblings"].append(sibling) 

352 self.collected_data["sibling_collected"] = collected + 1 

353 

354 if collected + 1 < count: 

355 return f"{collected + 2}人目の兄弟姉妹について教えてください。\n氏名、続柄、血縁タイプ、生年月日などを入力してください。" 

356 

357 # 全員収集完了 

358 self.state = InterviewState.SPECIAL_CASES 

359 self.collected_data["special_case_step"] = "renunciation" 

360 return "\n" + self.prompts.RENUNCIATION_QUESTION 

361 

362 def _process_special_cases(self, user_input: str) -> str: 

363 """特殊ケース（相続放棄等）を処理""" 

364 step = self.collected_data.get("special_case_step", "renunciation") 

365 

366 if step == "renunciation": 

367 has_renunciation = self._parse_yes_no(user_input) 

368 self.collected_data["has_renunciation"] = has_renunciation 

369 

370 # 次のステップへ 

371 self.collected_data["special_case_step"] = "retransfer" 

372 return "\n" + self.prompts.RETRANSFER_QUESTION 

373 

374 if step == "retransfer": 

375 has_retransfer = self._parse_yes_no(user_input) 

376 self.collected_data["has_retransfer"] = has_retransfer 

377 

378 # 確認へ 

379 self.state = InterviewState.CONFIRMATION 

380 summary = self._generate_summary() 

381 return "\n" + self.prompts.format_confirmation(summary) 

382 

383 return "エラーが発生しました。" 

384 

385 def _process_confirmation(self, user_input: str) -> str: 

386 """確認を処理""" 

387 confirmed = self._parse_yes_no(user_input) 

388 

389 if confirmed: 

390 self.state = InterviewState.COMPLETED 

391 return self.prompts.CALCULATION_START 

392 else: 

393 return "修正が必要な項目を教えてください。" 

394 

395 def _parse_yes_no(self, text: str) -> bool: 

396 """はい/いいえを解析""" 

397 text = text.strip().lower() 

398 if text in ["はい", "yes", "y", "有", "あり", "います", "いる"]: 

399 return True 

400 return False 

401 

402 def _extract_person_info(self, user_input: str, role: str) -> Dict[str, Any]: 

403 """ 

404 ユーザー入力から人物情報を抽出 

405 

406 Args: 

407 user_input: ユーザーの入力テキスト 

408 role: 役割（配偶者、子1、など） 

409 

410 Returns: 

411 抽出した人物情報の辞書 

412 """ 

413 # LLMを使って構造化データ抽出 

414 extraction_prompt = f""" 

415以下のテキストから{role}の情報を抽出してください。 

416抽出する情報: 

417- 氏名（name） 

418- 生存状態（is_alive: true/false） 

419- 生年月日（birth_date: YYYY-MM-DD形式、不明な場合はnull） 

420- 死亡日（death_date: YYYY-MM-DD形式、存命または不明な場合はnull） 

421- 性別（gender: male/female/other/unknown） 

422 

423テキスト: {user_input} 

424 

425以下のJSON形式で回答してください: 

426{{"name": "氏名", "is_alive": true, "birth_date": "YYYY-MM-DD", "death_date": null, "gender": "unknown"}} 

427""" 

428 

429 try: 

430 messages = [ 

431 {"role": "system", "content": "あなたは相続情報を正確に抽出するアシスタントです。"}, 

432 {"role": "user", "content": extraction_prompt} 

433 ] 

434 

435 response = self.client.chat(messages, temperature=0.1) 

436 

437 # JSON部分を抽出 

438 import json 

439 json_match = re.search(r'\{[^}]+\}', response) 

440 if json_match: 

441 extracted: Dict[str, Any] = json.loads(json_match.group()) 

442 

443 # 日付をdateオブジェクトに変換 

444 if extracted.get("birth_date"): 

445 extracted["birth_date"] = self._parse_date(extracted["birth_date"]) 

446 if extracted.get("death_date"): 

447 extracted["death_date"] = self._parse_date(extracted["death_date"]) 

448 

449 # 性別をGenderに変換 

450 if extracted.get("gender"): 

451 gender_map = { 

452 "male": Gender.MALE, 

453 "female": Gender.FEMALE, 

454 "other": Gender.OTHER, 

455 "unknown": Gender.UNKNOWN 

456 } 

457 extracted["gender"] = gender_map.get(extracted["gender"].lower(), Gender.UNKNOWN) 

458 

459 return extracted 

460 

461 except Exception as e: 

462 self.logger.warning(f"Failed to extract person info with LLM: {e}") 

463 

464 # フォールバック: 簡易パース 

465 return self._simple_parse_person_info(user_input, role) 

466 

467 def _simple_parse_person_info(self, user_input: str, role: str) -> Dict[str, Any]: 

468 """ 

469 簡易的な人物情報パース（フォールバック用） 

470 

471 Args: 

472 user_input: ユーザー入力 

473 role: 役割 

474 

475 Returns: 

476 抽出した情報 

477 """ 

478 lines = user_input.strip().split("\n") 

479 name = lines[0] if lines else role 

480 

481 # 生存状態の推測 

482 is_alive = "死亡" not in user_input and "亡くなっ" not in user_input 

483 

484 # 日付の抽出 

485 birth_date = None 

486 death_date = None 

487 for line in lines: 

488 if "生年月日" in line or "生まれ" in line: 

489 birth_date = self._parse_date(line) 

490 if "死亡" in line or "亡くなっ" in line: 

491 death_date = self._parse_date(line) 

492 is_alive = False 

493 

494 return { 

495 "name": name, 

496 "is_alive": is_alive, 

497 "birth_date": birth_date, 

498 "death_date": death_date, 

499 "gender": Gender.UNKNOWN 

500 } 

501 

502 def _parse_date(self, text: str) -> Optional[date]: 

503 """日付文字列を解析""" 

504 text = text.strip() 

505 

506 # YYYY-MM-DD形式 

507 match = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', text) 

508 if match: 

509 try: 

510 return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) 

511 except ValueError: 

512 pass 

513 

514 # YYYY/MM/DD形式 

515 match = re.match(r'(\d{4})/(\d{1,2})/(\d{1,2})', text) 

516 if match: 

517 try: 

518 return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) 

519 except ValueError: 

520 pass 

521 

522 # YYYY年MM月DD日形式 

523 match = re.match(r'(\d{4})年(\d{1,2})月(\d{1,2})日', text) 

524 if match: 

525 try: 

526 return date(int(match.group(1)), int(match.group(2)), int(match.group(3))) 

527 except ValueError: 

528 pass 

529 

530 return None 

531 

532 def _generate_summary(self) -> str: 

533 """収集した情報のサマリーを生成""" 

534 lines = [] 

535 lines.append(f"被相続人: {self.decedent.name if self.decedent else '不明'}") 

536 if self.decedent and self.decedent.death_date: 

537 lines.append(f"死亡日: {self.decedent.death_date}") 

538 

539 if self.spouses: 

540 lines.append(f"\n配偶者: {', '.join(s.name for s in self.spouses)}") 

541 

542 if self.children: 

543 lines.append(f"\n子: {', '.join(c.name for c in self.children)}") 

544 

545 if self.parents: 

546 lines.append(f"\n直系尊属: {', '.join(p.name for p in self.parents)}") 

547 

548 if self.siblings: 

549 lines.append(f"\n兄弟姉妹: {', '.join(s.name for s in self.siblings)}") 

550 

551 return "\n".join(lines) 

552 

553 def get_collected_data(self) -> Dict[str, Any]: 

554 """ 

555 収集したデータを取得 

556 

557 Returns: 

558 相続計算に必要なデータ 

559 """ 

560 return { 

561 "decedent": self.decedent, 

562 "spouses": self.spouses, 

563 "children": self.children, 

564 "parents": self.parents, 

565 "siblings": self.siblings, 

566 "renounced": self.renounced, 

567 "disqualified": self.disqualified, 

568 "disinherited": self.disinherited, 

569 "sibling_blood_types": self.sibling_blood_types, 

570 "retransfer_heirs_info": self.retransfer_heirs_info, 

571 } 

572 

573 def is_completed(self) -> bool: 

574 """インタビューが完了したか確認""" 

575 return self.state == InterviewState.COMPLETED