from abc import ABC, abstractmethod
from typing import Any, List, Optional, Dict
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, SystemMessage, AIMessageChunk
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableConfig
from tools.exception import InterruptedException
from Agents.llms.LlmInterface import LLMInterface # Import the interface
try:
# ollamaライブラリがインストールされている場合、専用のエラーをインポート
from ollama import ResponseError
except ImportError:
ResponseError = None # インストールされていない場合はNoneにしておく
class LlmBase(LLMInterface): # LlmBase implements LLMInterface
model_name: str
temperature: float
llm_kwargs: Dict[str, Any]
_supports_images: bool
llm: BaseChatModel # The actual Langchain LLM instance
def __init__(self, model_identifier: str, temperature: float = 0, **kwargs):
self.model_name = model_identifier
self.temperature = temperature
self.llm_kwargs = kwargs
self.tools: List[Any] = []
self.system_prompt: str = ""
self._supports_images = False # Default, to be overridden by subclasses
self.llm = self._initialize_llm() # Initialize the specific LLM in the constructor
self.chat_history = None
self.private_memory = False
self.name=""
self. agent = self.create_character()
@abstractmethod
def _initialize_llm(self) -> BaseChatModel:
"""Abstract method to initialize the specific Langchain LLM instance."""
pass
def get_langchain_llm_instance(self) -> Optional[BaseChatModel]:
return self.llm
#######################
def create_character(self):
if self.private_memory:
print("self.system_prompt", self.system_prompt)
print("self.name", self.name)
self.prompt = ChatPromptTemplate.from_messages([
("system", self.system_prompt),
MessagesPlaceholder(variable_name=self.name + "chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
else:
if 0 < len(self.system_prompt):
self.prompt = ChatPromptTemplate.from_messages([
("system", self.system_prompt),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
else:
self.prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
return self.create_agent_executer()
def create_agent_executer(self):
#各種設定が行われたときに作り直す。get_responseでは作らない。連続してメッセージをやり取りするときの負荷低減
agent = create_tool_calling_agent(self.llm, self.tools, self.prompt)
self.agent = AgentExecutor(
agent=agent,
tools=self.tools,
verbose=True,
memory=self.chat_history
)
return self.agent
###################
def update_tools(self, tools: List[Any]):
"""Updates the list of tools available to the LLM provider."""
self.tools = tools
def update_system_prompt(self, system_prompt: str):
"""Updates the system prompt for the LLM provider."""
self.system_prompt_str = system_prompt
def get_current_message_content(self,prompt,image_paths):
# Construct the HumanMessage content, handling multimodal input
current_message_content: Any = prompt
if self.supports_images and image_paths:
content_parts = [{"type": "text", "text": prompt}]
for img_data_url in image_paths:
# Assuming image_paths already contain base64 data URLs from AIAgent
content_parts.append({
"type": "image_url",
"image_url": {"url": img_data_url}
})
current_message_content = content_parts
return current_message_content
def get_response(self,
prompt: str,
chat_history: Optional[List[BaseMessage]] = None,
image_paths: Optional[List[str]] = None,
system_prompt: Optional[str] = None,
tools: Optional[List[Any]] = None,
callbacks: Optional[List[Any]] = None,
**kwargs) -> str:
# Construct the HumanMessage content, handling multimodal input
current_message_content = self.get_current_message_content(prompt, image_paths)
run_config = RunnableConfig(callbacks=callbacks if callbacks else [])
# ツールが指定されている場合のみAgentExecutorを使用
try:
agent_input = {"input": current_message_content, "chat_history": chat_history if chat_history else []}
response = self.agent.invoke(agent_input, config=run_config, **kwargs)
# ★★★【重要】成功した場合の戻り値を返す処理を追加 ★★★
if isinstance(response, dict):
return response.get("output", f"AgentExecutor returned a dict without 'output' key: {response}")
return str(response)
except InterruptedException:
print(f"{type(self).__name__} ({self.model_name}): AgentExecutor execution interrupted.")
raise # AIAgent側で処理するために再スロー
except NotImplementedError as e_not_implemented:
# bind_tools が実装されていないモデルでツールを使おうとした場合のエラー
import traceback
error_msg = f"{type(self).__name__} ({self.model_name}) error: The selected model or its LangChain wrapper does not support the required tool-calling feature (bind_tools). Please use a tool-compatible model (like `devstral` with the latest `langchain-ollama` package) or disable tools. Original error: {e_not_implemented}"
print(error_msg)
traceback.print_exc()
return error_msg
except ResponseError as e:
error_msg = f"Ollama server returned an error: {e}"
print(error_msg)
# このエラーもGUIに返す
return f"{type(self).__name__} AgentExecutor error: {error_msg}"
except Exception as e_agent:
import traceback
print(f"{type(self).__name__}: Error during AgentExecutor execution: {e_agent}")
traceback.print_exc() # デバッグ用に詳細なエラー情報を出力
return f"{type(self).__name__} AgentExecutor error: {e_agent}"
except BaseException as e_agent:
import traceback
print(f"{type(self).__name__}: Error during AgentExecutor execution: {e_agent}")
traceback.print_exc() # デバッグ用に詳細なエラー情報を出力
return f"{type(self).__name__} AgentExecutor error: {e_agent}"
@property
@abstractmethod # This must be abstract in LlmBase as it's specific to each LLM
def supports_images(self) -> bool:
passAIによる説明
このコードは、様々な大規模言語モデル(LLM)を統一的に扱うための基底クラスLlmBaseと、その関連関数群を定義しています。Langchainライブラリをベースに、ツール呼び出し機能を持つエージェントを構築し、テキストだけでなく画像入力にも対応できるように設計されています。
クラス LlmBase
LlmBaseは、LLMInterfaceというインターフェースを実装した抽象クラスです。具体的なLLM（例えば、Ollama、Llama2など）の実装は、LlmBaseを継承したサブクラスで行われます。 LlmBase自身は、LLMの初期化やエージェントの作成、応答取得といった共通の機能を提供します。
model_name: str: 使用するLLMモデルの名前を表す文字列。temperature: float: 応答のランダム性を制御するパラメータ。値が大きいほどランダムな応答になります。llm_kwargs: Dict[str, Any]: LLMの初期化に渡す追加のキーワード引数。_supports_images: bool: LLMが画像入力をサポートするかどうかを示すブール値。サブクラスでオーバーライドされます。llm: BaseChatModel: LangchainのBaseChatModelオブジェクト。具体的なLLMインスタンスを保持します。tools: List[Any]: LLMが利用できるツールのリスト。system_prompt: str: LLMへのシステムプロンプト（初期指示）。chat_history: Optional[List[BaseMessage]]: チャット履歴。private_memory: bool: プライベートメモリを使用するかどうか。name: str: エージェントの名前。agent: AgentExecutor: ツール呼び出しエージェント。
メソッド
__init__(self, model_identifier: str, temperature: float = 0, **kwargs): コンストラクタ。model_identifierでLLMを特定し、temperatureとkwargsでLLMを初期化します。_initialize_llm()を呼び出して具体的なLLMインスタンスを作成します。_initialize_llm(self) -> BaseChatModel: 抽象メソッド。サブクラスで実装され、具体的なLangchain LLMインスタンスを初期化して返します。get_langchain_llm_instance(self) -> Optional[BaseChatModel]: LangchainのLLMインスタンスを返します。create_character(self): システムプロンプトとチャット履歴に基づいて、LangchainのAgentExecutorを作成します。private_memoryフラグに応じて、プライベートメモリを使用するかどうかを決定します。create_agent_executer(self):AgentExecutorを生成します。ツールやプロンプトの設定が変更された際に呼び出され、エージェントを再構築します。update_tools(self, tools: List[Any]): LLMが利用できるツールのリストを更新します。update_system_prompt(self, system_prompt: str): システムプロンプトを更新します。get_current_message_content(self, prompt, image_paths): テキストプロンプトと画像パスを受け取り、Langchainが処理できる形式のメッセージコンテンツを作成します。画像データはbase64エンコードされたURLとして扱われます。get_response(self, prompt: str, ...): ユーザーからのプロンプトを受け取り、LLMに質問し、応答を取得します。chat_history,image_paths,system_prompt,tools,callbacksなどの引数を受け付け、柔軟な制御を可能にしています。エラー処理も充実しており、様々な例外をキャッチして適切なエラーメッセージを返します。supports_images(self) -> bool: 抽象メソッド。LLMが画像入力をサポートするかどうかを返します。サブクラスで実装する必要があります。
エラー処理
get_responseメソッドでは、InterruptedException, NotImplementedError, ResponseError, Exception, BaseExceptionといった様々な例外をキャッチし、適切なエラーメッセージを返しています。これにより、LLMとの通信エラーや、モデルの機能不足などによるエラーをユーザーに分かりやすく伝えることができます。
全体的な設計
このコードは、LLMの具体的な実装を抽象化し、共通のインターフェースを提供することで、様々なLLMを容易に交換できるように設計されています。また、ツール呼び出し機能や画像入力への対応など、高度な機能も備えています。エラー処理も充実しており、堅牢なシステム構築に役立ちます。 create_agent_executer関数の存在は、パフォーマンスの最適化を意図していると考えられます。連続したメッセージ処理において、エージェントの再生成を避けることで効率性を向上させています。