my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import os
#2	from typing import Any, Dict, List, Optional
#3
#4	from mem0.configs.llms.base import BaseLlmConfig
#5
#6
#7	class AWSBedrockConfig(BaseLlmConfig):
#8	"""
#9	Configuration class for AWS Bedrock LLM integration.
#10
#11	Supports all available Bedrock models with automatic provider detection.
#12	"""
#13
#14	def __init__(
#15	self,
#16	model: Optional[str] = None,
#17	temperature: float = 0.1,
#18	max_tokens: int = 2000,
#19	top_p: float = 0.9,
#20	top_k: int = 1,
#21	aws_access_key_id: Optional[str] = None,
#22	aws_secret_access_key: Optional[str] = None,
#23	aws_region: str = "",
#24	aws_session_token: Optional[str] = None,
#25	aws_profile: Optional[str] = None,
#26	model_kwargs: Optional[Dict[str, Any]] = None,
#27	**kwargs,
#28	):
#29	"""
#30	Initialize AWS Bedrock configuration.
#31
#32	Args:
#33	model: Bedrock model identifier (e.g., "amazon.nova-3-mini-20241119-v1:0")
#34	temperature: Controls randomness (0.0 to 2.0)
#35	max_tokens: Maximum tokens to generate
#36	top_p: Nucleus sampling parameter (0.0 to 1.0)
#37	top_k: Top-k sampling parameter (1 to 40)
#38	aws_access_key_id: AWS access key (optional, uses env vars if not provided)
#39	aws_secret_access_key: AWS secret key (optional, uses env vars if not provided)
#40	aws_region: AWS region for Bedrock service
#41	aws_session_token: AWS session token for temporary credentials
#42	aws_profile: AWS profile name for credentials
#43	model_kwargs: Additional model-specific parameters
#44	**kwargs: Additional arguments passed to base class
#45	"""
#46	super().__init__(
#47	model=model or "anthropic.claude-3-5-sonnet-20240620-v1:0",
#48	temperature=temperature,
#49	max_tokens=max_tokens,
#50	top_p=top_p,
#51	top_k=top_k,
#52	**kwargs,
#53	)
#54
#55	self.aws_access_key_id = aws_access_key_id
#56	self.aws_secret_access_key = aws_secret_access_key
#57	self.aws_region = aws_region or os.getenv("AWS_REGION", "us-west-2")
#58	self.aws_session_token = aws_session_token
#59	self.aws_profile = aws_profile
#60	self.model_kwargs = model_kwargs or {}
#61
#62	@property
#63	def provider(self) -> str:
#64	"""Get the provider from the model identifier."""
#65	if not self.model or "." not in self.model:
#66	return "unknown"
#67	return self.model.split(".")[0]
#68
#69	@property
#70	def model_name(self) -> str:
#71	"""Get the model name without provider prefix."""
#72	if not self.model or "." not in self.model:
#73	return self.model
#74	return ".".join(self.model.split(".")[1:])
#75
#76	def get_model_config(self) -> Dict[str, Any]:
#77	"""Get model-specific configuration parameters."""
#78	base_config = {
#79	"temperature": self.temperature,
#80	"max_tokens": self.max_tokens,
#81	"top_p": self.top_p,
#82	"top_k": self.top_k,
#83	}
#84
#85	# Add custom model kwargs
#86	base_config.update(self.model_kwargs)
#87
#88	return base_config
#89
#90	def get_aws_config(self) -> Dict[str, Any]:
#91	"""Get AWS configuration parameters."""
#92	config = {
#93	"region_name": self.aws_region,
#94	}
#95
#96	if self.aws_access_key_id:
#97	config["aws_access_key_id"] = self.aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
#98
#99	if self.aws_secret_access_key:
#100	config["aws_secret_access_key"] = self.aws_secret_access_key or os.getenv("AWS_SECRET_ACCESS_KEY")
#101
#102	if self.aws_session_token:
#103	config["aws_session_token"] = self.aws_session_token or os.getenv("AWS_SESSION_TOKEN")
#104
#105	if self.aws_profile:
#106	config["profile_name"] = self.aws_profile or os.getenv("AWS_PROFILE")
#107
#108	return config
#109
#110	def validate_model_format(self) -> bool:
#111	"""
#112	Validate that the model identifier follows Bedrock naming convention.
#113
#114	Returns:
#115	True if valid, False otherwise
#116	"""
#117	if not self.model:
#118	return False
#119
#120	# Check if model follows provider.model-name format
#121	if "." not in self.model:
#122	return False
#123
#124	provider, model_name = self.model.split(".", 1)
#125
#126	# Validate provider
#127	valid_providers = [
#128	"ai21", "amazon", "anthropic", "cohere", "meta", "mistral",
#129	"stability", "writer", "deepseek", "gpt-oss", "perplexity",
#130	"snowflake", "titan", "command", "j2", "llama"
#131	]
#132
#133	if provider not in valid_providers:
#134	return False
#135
#136	# Validate model name is not empty
#137	if not model_name:
#138	return False
#139
#140	return True
#141
#142	def get_supported_regions(self) -> List[str]:
#143	"""Get list of AWS regions that support Bedrock."""
#144	return [
#145	"us-east-1",
#146	"us-west-2",
#147	"us-east-2",
#148	"eu-west-1",
#149	"ap-southeast-1",
#150	"ap-northeast-1",
#151	]
#152
#153	def get_model_capabilities(self) -> Dict[str, Any]:
#154	"""Get model capabilities based on provider."""
#155	capabilities = {
#156	"supports_tools": False,
#157	"supports_vision": False,
#158	"supports_streaming": False,
#159	"supports_multimodal": False,
#160	}
#161
#162	if self.provider == "anthropic":
#163	capabilities.update({
#164	"supports_tools": True,
#165	"supports_vision": True,
#166	"supports_streaming": True,
#167	"supports_multimodal": True,
#168	})
#169	elif self.provider == "amazon":
#170	capabilities.update({
#171	"supports_tools": True,
#172	"supports_vision": True,
#173	"supports_streaming": True,
#174	"supports_multimodal": True,
#175	})
#176	elif self.provider == "cohere":
#177	capabilities.update({
#178	"supports_tools": True,
#179	"supports_streaming": True,
#180	})
#181	elif self.provider == "meta":
#182	capabilities.update({
#183	"supports_vision": True,
#184	"supports_streaming": True,
#185	})
#186	elif self.provider == "mistral":
#187	capabilities.update({
#188	"supports_vision": True,
#189	"supports_streaming": True,
#190	})
#191
#192	return capabilities
#193

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public