This section explains how the framework integrates with large language models (LLMs) through the llm_provider.py module. The LLM integration is built around an abstract base class and concrete provider implementations that facilitate communication with various LLM APIs.
LLM Integration
Overview
1. LLMProvider (Abstract Base Class)
Purpose
The LLMProvider class defines a common interface that all LLM providers must implement. This abstraction ensures that the agent can switch between different providers (e.g., OpenAI, Anthropic, or custom providers) with minimal changes to the core agent logic.
Key Methods
-
format_tools(tools: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]
Converts the tool configuration (loaded from JSON files) into the format required by the specific LLM API. For example, OpenAI's function calling requires a structured JSON format. -
get_response(messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]]) -> str
Sends the conversation messages (including system prompt, user messages, and memory) along with optional tool information to the LLM and returns the model's response. -
extract_tool_call(response_text: str) -> Optional[Dict[str, Any]]
Parses the LLM's response to detect if it includes a tool call, typically formatted as JSON. This method extracts the tool name and parameters if present.
Example (Abstract Behavior)
class LLMProvider:
def format_tools(self, tools: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
raise NotImplementedError("Subclasses must implement format_tools")
def get_response(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]]) -> str:
raise NotImplementedError("Subclasses must implement get_response")
def extract_tool_call(self, response_text: str) -> Optional[Dict[str, Any]]:
# Common logic to search for JSON formatted tool calls in the response.
try:
import json, re
json_match = re.search(r'```(?:json)?\s*({[\s\S]*?})\s*```', response_text, re.DOTALL)
if not json_match:
json_match = re.search(r'^({[\s\S]*})$', response_text.strip(), re.DOTALL)
if json_match:
tool_call = json.loads(json_match.group(1))
if "tool" in tool_call and "parameters" in tool_call:
return tool_call
except Exception as e:
print(f"Error extracting tool call: {str(e)}")
return None2. OpenAIProvider (Concrete Implementation)
Purpose
Implements the LLMProvider interface for OpenAI's GPT models, utilizing their function-calling feature.
Key Features
-
Tool Formatting:
Converts tool definitions into a JSON schema that the OpenAI API expects for function calling. -
Response Handling:
Handles communication with the OpenAI API, including error handling if the API key is missing or if the module isn't installed. -
Function Extraction:
Uses a regular expression to extract a JSON block from the response, which is expected to include a tool call when the LLM suggests using a tool.
Example Code
class OpenAIProvider(LLMProvider):
def __init__(self, model="gpt-4"):
self.model = model
self.api_key = os.environ.get("OPENAI_API_KEY")
if not self.api_key:
print("Warning: OPENAI_API_KEY not set in environment variables")
def format_tools(self, tools: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
formatted_tools = []
for tool_name, tool_config in tools.items():
formatted_tool = {
"type": "function",
"function": {
"name": tool_name,
"description": tool_config.get("description", ""),
"parameters": tool_config.get("parameters", {})
}
}
formatted_tools.append(formatted_tool)
return formatted_tools
def get_response(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]]) -> str:
if not self.api_key:
return "Error: OpenAI API key not set. Please set the OPENAI_API_KEY environment variable."
try:
import importlib.util
if importlib.util.find_spec("openai") is None:
return "Error: OpenAI module not installed. Please install it with 'pip install openai'"
import openai
client = openai.Client(api_key=self.api_key)
args = {
"model": self.model,
"messages": messages,
"temperature": 0.7
}
if tools:
args["tools"] = tools
args["tool_choice"] = "auto"
response = client.chat.completions.create(**args)
return response.choices[0].message.content
except Exception as e:
return f"Error calling OpenAI API: {str(e)}"3. AnthropicProvider (Concrete Implementation)
Purpose
Provides integration with Anthropic's Claude model. Since Anthropic's API handles tool information differently (often embedded in the system prompt), this provider doesn't format tool calls as directly as OpenAI.
Key Features
-
System Prompt Handling:
Extracts and prepares system messages differently, as Claude requires a specific format. -
Response Parsing:
Processes Claude's response to extract content, handling nuances of Anthropic's message structure.
Example Code
class AnthropicProvider(LLMProvider):
def __init__(self, model="claude-3-opus-20240229"):
self.model = model
self.api_key = os.environ.get("ANTHROPIC_API_KEY")
if not self.api_key:
print("Warning: ANTHROPIC_API_KEY not set in environment variables")
def format_tools(self, tools: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
# Claude uses prompt-based tool integration rather than function calls.
return []
def get_response(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]]) -> str:
if not self.api_key:
return "Error: Anthropic API key not set. Please set the ANTHROPIC_API_KEY environment variable."
try:
import importlib.util
if importlib.util.find_spec("anthropic") is None:
return "Error: Anthropic module not installed. Please install it with 'pip install anthropic'"
import anthropic
client = anthropic.Anthropic(api_key=self.api_key)
system_prompt = ""
claude_messages = []
for msg in messages:
if msg["role"] == "system":
if not claude_messages:
system_prompt = msg["content"]
else:
claude_messages.append({
"role": "assistant",
"content": f"System message: {msg['content']}"
})
elif msg["role"] == "user":
claude_messages.append({"role": "user", "content": msg["content"]})
elif msg["role"] == "assistant":
claude_messages.append({"role": "assistant", "content": msg["content"]})
response = client.messages.create(
model=self.model,
system=system_prompt,
messages=claude_messages,
max_tokens=2000,
temperature=0.7
)
return response.content[0].text
except Exception as e:
return f"Error calling Anthropic API: {str(e)}"4. Custom LLM Provider (Example Extension)
If you want to integrate an LLM provider that isn't natively supported, you can create a custom provider by extending LLMProvider.
Example Code
class CustomLLMProvider(LLMProvider):
"""Example custom provider implementation using a hypothetical API."""
def __init__(self, model="custom-model"):
self.model = model
self.api_key = os.environ.get("CUSTOM_API_KEY")
if not self.api_key:
print("Warning: CUSTOM_API_KEY not set in environment variables")
def format_tools(self, tools: Dict[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
formatted_tools = []
for tool_name, tool_config in tools.items():
formatted_tool = {
"type": "function",
"function": {
"name": tool_name,
"description": tool_config.get("description", ""),
"parameters": tool_config.get("parameters", {})
}
}
formatted_tools.append(formatted_tool)
return formatted_tools
def get_response(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]]) -> str:
# Replace this stub with your API call logic.
return "Response from CustomLLMProvider."
def extract_tool_call(self, response_text: str) -> Optional[Dict[str, Any]]:
return NoneTo use your custom provider, update the provider selector in llm_provider.py:
def get_llm_provider(provider_name: str) -> LLMProvider:
if provider_name.lower() == "openai":
return OpenAIProvider()
elif provider_name.lower() == "anthropic":
return AnthropicProvider()
elif provider_name.lower() == "custom":
return CustomLLMProvider()
else:
raise ValueError(f"Unsupported LLM provider: {provider_name}")Summary
-
Abstraction:
TheLLMProviderabstract class standardizes how LLMs are integrated. -
Concrete Implementations:
Providers likeOpenAIProviderandAnthropicProviderimplement specific API interactions. -
Flexibility:
The design allows easy integration of new providers through a common interface. -
Tool Call Extraction:
Theextract_tool_callmethod scans LLM responses for JSON blocks that indicate a tool call, enabling dynamic execution of external functions.
By leveraging this design, the framework provides a flexible, robust, and extensible way to interact with multiple LLM APIs, allowing you to adapt to different environments and requirements with minimal changes to your core agent logic.