Spaces:
Running
Running
File size: 5,071 Bytes
6a71f13 fac50d3 6a71f13 227e75d 6a71f13 fac50d3 227e75d 6a71f13 0520d98 6a71f13 227e75d 6a71f13 227e75d 6a71f13 227e75d 6a71f13 227e75d 8969ec5 6a71f13 e32a461 8969ec5 6a71f13 8969ec5 6a71f13 227e75d 6a71f13 8969ec5 227e75d 8baa940 |
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 |
from typing import Optional, List, Dict, Any
from openai import OpenAI
import anthropic
from dataclasses import dataclass
from config.llm_settings import LLMSettings
from core.file_scanner import FileInfo
@dataclass
class Message:
role: str
content: str
class LLMService:
MAX_TURNS = 5
def __init__(self):
"""LLMサービスの初期化"""
self.settings = LLMSettings()
self.current_model = self.settings.default_llm
# API クライアントの初期化
if self.settings.anthropic_api_key:
self.claude_client = anthropic.Anthropic(api_key=self.settings.anthropic_api_key)
if self.settings.openai_api_key:
self.openai_client = OpenAI(api_key=self.settings.openai_api_key)
self.conversation_history: List[Message] = []
def switch_model(self, model: str):
"""使用するモデルを切り替え"""
model_name = model.lower()
if model_name == "claude" and self.settings.anthropic_api_key:
self.current_model = "claude"
elif model_name == "openai" and self.settings.openai_api_key:
self.current_model = "openai"
else:
available = []
if self.settings.anthropic_api_key:
available.append("claude")
if self.settings.openai_api_key:
available.append("openai")
raise ValueError(f"利用可能なモデル: {', '.join(available)}")
def create_prompt(self, content: str, query: str) -> str:
"""プロンプトを生成"""
return f"""以下はGitHubリポジトリのコード解析結果です。このコードについて質問に答えてください。
コード解析結果:
{content}
質問: {query}
できるだけ具体的に、コードの内容を参照しながら回答してください。"""
def _add_to_history(self, role: str, content: str):
"""会話履歴に追加（最大5ターン）"""
self.conversation_history.append(Message(role=role, content=content))
# 最大ターン数を超えた場合、古い会話を削除
if len(self.conversation_history) > self.MAX_TURNS * 2: # 各ターンは質問と回答で2メッセージ
self.conversation_history = self.conversation_history[-self.MAX_TURNS * 2:]
def _format_messages_for_claude(self) -> List[Dict[str, str]]:
"""Claude用にメッセージをフォーマット"""
return [{"role": msg.role, "content": msg.content}
for msg in self.conversation_history]
def _format_messages_for_gpt(self) -> List[Dict[str, str]]:
"""GPT用にメッセージをフォーマット"""
return [
{"role": "system", "content": "あなたはコードアナリストとして、リポジトリの解析と質問への回答を行います。"},
*[{"role": msg.role, "content": msg.content}
for msg in self.conversation_history]
]
def get_conversation_history(self) -> List[Dict[str, str]]:
"""会話履歴を取得"""
return [{"role": msg.role, "content": msg.content}
for msg in self.conversation_history]
def clear_history(self):
"""会話履歴をクリア"""
self.conversation_history = []
def get_response(self, content: str, query: str) -> tuple[Optional[str], Optional[str]]:
"""LLMを使用して回答を生成"""
try:
prompt = self.create_prompt(content, query)
self._add_to_history("user", prompt)
print(f"Current model: {self.current_model}") # デバッグ用出力
if self.current_model.lower() == 'claude':
print("Using Claude API") # デバッグ用出力
response = self.claude_client.messages.create(
model="claude-3-5-sonnet-latest",
messages=self._format_messages_for_claude()
)
answer = response.content[0].text
else: # gpt
print("Using GPT API") # デバッグ用出力
response = openai.chat.completions.create(
model="gpt-4",
messages=self._format_messages_for_gpt()
)
answer = response.choices[0].message.content
self._add_to_history("assistant", answer)
return answer, None
except Exception as e:
return None, f"エラーが発生しました: {str(e)}"
@staticmethod
def format_code_content(files: List[FileInfo]) -> str:
"""ファイル内容をプロンプト用にフォーマット"""
formatted_content = []
for file_info in files:
formatted_content.append(
f"#ファイルパス\n{file_info.path}\n------------\n{file_info.content}\n"
)
return "\n".join(formatted_content) |