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

1"""Ollamaクライアント 

2 

3Ollamaとの通信を管理し、LLMとのインタラクションを提供する。 

4""" 

5from typing import Optional, Dict, Any, List 

6import logging 

7 

8import ollama 

9from ollama import ChatResponse 

10 

11from ..utils.config import Settings 

12from ..utils.exceptions import ServiceException 

13 

14 

15class OllamaClient: 

16 """ 

17 Ollamaクライアント 

18 

19 gpt-oss:20bモデルとの通信を管理し、 

20 相続に関する質問応答機能を提供する。 

21 """ 

22 

23 def __init__( 

24 self, 

25 model: str = "gpt-oss:20b", 

26 host: Optional[str] = None, 

27 timeout: int = 120 

28 ) -> None: 

29 """ 

30 初期化 

31 

32 Args: 

33 model: 使用するモデル名 

34 host: OllamaホストURL（Noneの場合は設定から取得） 

35 timeout: タイムアウト時間（秒） 

36 """ 

37 self.logger = logging.getLogger(__name__) 

38 

39 # 設定の取得（Ollamaのみを初期化） 

40 if host is None: 

41 from ..utils.config import OllamaSettings 

42 ollama_settings = OllamaSettings() 

43 host = ollama_settings.host 

44 

45 self.model = model 

46 self.host = host 

47 self.timeout = timeout 

48 

49 # クライアントの初期化確認 

50 self._verify_connection() 

51 

52 self.logger.info( 

53 f"OllamaClient initialized: model={self.model}, host={self.host}" 

54 ) 

55 

56 def _verify_connection(self) -> None: 

57 """ 

58 Ollamaサーバーへの接続を確認 

59 

60 Raises: 

61 ServiceException: 接続に失敗した場合 

62 """ 

63 try: 

64 # モデルリストの取得で接続確認 

65 models = ollama.list() 

66 

67 # モデルの存在確認 

68 model_names = [m.model for m in models.models if m.model is not None] 

69 if self.model not in model_names: 

70 available = ", ".join(model_names) 

71 raise ServiceException( 

72 f"Model '{self.model}' not found. " 

73 f"Available models: {available}" 

74 ) 

75 

76 self.logger.info(f"Connected to Ollama successfully") 

77 

78 except Exception as e: 

79 error_msg = f"Failed to connect to Ollama: {str(e)}" 

80 self.logger.error(error_msg) 

81 raise ServiceException(error_msg) from e 

82 

83 def chat( 

84 self, 

85 messages: List[Dict[str, str]], 

86 temperature: float = 0.7, 

87 max_tokens: Optional[int] = None 

88 ) -> str: 

89 """ 

90 チャット形式でLLMと対話 

91 

92 Args: 

93 messages: メッセージリスト（[{"role": "user", "content": "..."}]形式） 

94 temperature: 生成のランダム性（0.0-1.0） 

95 max_tokens: 最大トークン数 

96 

97 Returns: 

98 LLMの応答テキスト 

99 

100 Raises: 

101 ServiceException: 通信エラーが発生した場合 

102 """ 

103 try: 

104 self.logger.debug(f"Sending chat request with {len(messages)} messages") 

105 

106 # Ollamaのchat APIを呼び出し 

107 options: Dict[str, Any] = { 

108 "temperature": temperature, 

109 } 

110 if max_tokens: 

111 options["num_predict"] = max_tokens 

112 

113 response: ChatResponse = ollama.chat( 

114 model=self.model, 

115 messages=messages, 

116 options=options 

117 ) 

118 

119 # 応答の取得 

120 content = response.message.content 

121 if content is None: 

122 self.logger.warning("Received None content from Ollama") 

123 return "" 

124 

125 self.logger.debug(f"Received response: {len(content)} characters") 

126 

127 return content 

128 

129 except Exception as e: 

130 error_msg = f"Chat request failed: {str(e)}" 

131 self.logger.error(error_msg) 

132 raise ServiceException(error_msg) from e 

133 

134 def generate( 

135 self, 

136 prompt: str, 

137 system: Optional[str] = None, 

138 temperature: float = 0.7, 

139 max_tokens: Optional[int] = None 

140 ) -> str: 

141 """ 

142 プロンプトから文章を生成 

143 

144 Args: 

145 prompt: プロンプトテキスト 

146 system: システムプロンプト（オプション） 

147 temperature: 生成のランダム性（0.0-1.0） 

148 max_tokens: 最大トークン数 

149 

150 Returns: 

151 生成されたテキスト 

152 

153 Raises: 

154 ServiceException: 通信エラーが発生した場合 

155 """ 

156 messages: List[Dict[str, str]] = [] 

157 

158 if system: 

159 messages.append({"role": "system", "content": system}) 

160 

161 messages.append({"role": "user", "content": prompt}) 

162 

163 return self.chat( 

164 messages=messages, 

165 temperature=temperature, 

166 max_tokens=max_tokens 

167 ) 

168 

169 def ask_question( 

170 self, 

171 question: str, 

172 context: Optional[str] = None 

173 ) -> str: 

174 """ 

175 質問に対する回答を生成 

176 

177 Args: 

178 question: 質問テキスト 

179 context: 質問のコンテキスト（会話履歴など） 

180 

181 Returns: 

182 回答テキスト 

183 

184 Raises: 

185 ServiceException: 通信エラーが発生した場合 

186 """ 

187 system_prompt = """あなたは日本の相続法に詳しい専門家です。 

188被相続人の相続に関する情報を収集するために、ユーザーに質問をします。 

189回答は簡潔で分かりやすく、必要に応じて法的な説明を加えてください。""" 

190 

191 if context: 

192 prompt = f"コンテキスト: {context}\n\n質問: {question}" 

193 else: 

194 prompt = question 

195 

196 return self.generate( 

197 prompt=prompt, 

198 system=system_prompt, 

199 temperature=0.3 # 法的情報なので低めの温度 

200 ) 

201 

202 def parse_user_input( 

203 self, 

204 user_input: str, 

205 expected_format: str 

206 ) -> str: 

207 """ 

208 ユーザー入力を解析して構造化 

209 

210 Args: 

211 user_input: ユーザーの入力 

212 expected_format: 期待される形式の説明 

213 

214 Returns: 

215 解析結果（JSON形式の文字列など） 

216 

217 Raises: 

218 ServiceException: 解析エラーが発生した場合 

219 """ 

220 system_prompt = f"""ユーザーの入力を解析して、以下の形式で出力してください。 

221出力形式: {expected_format} 

222 

223余計な説明は不要です。指定された形式のみを出力してください。""" 

224 

225 return self.generate( 

226 prompt=user_input, 

227 system=system_prompt, 

228 temperature=0.1 # 解析タスクなので低温度 

229 )