Appearance
SageMakerAIModel
Invoke a deployed SageMaker endpoint. Requires the sagemaker extra and AWS credentials:
bash
pip install --extra-index-url https://core-packages.elsai.ai/root/ "elsai-model[sagemaker]==2.1.0"Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, SAGEMAKER_ENDPOINT_NAME
python
import os
from elsai_model.sagemaker import SageMakerAIModel
model = SageMakerAIModel(
endpoint_config={
"endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
"region_name": os.getenv("AWS_REGION", "us-west-2"),
},
payload_config={"max_tokens": 256, "temperature": 0.2, "stream": False},
)
messages = [{"role": "user", "content": "Say hello in one short sentence."}]
response = model.invoke(messages)
print(response["choices"][0]["message"]["content"])Or via the factory (no model=; use endpoint + payload config):
python
from elsai_model import LLM, Provider
model = LLM(
provider=Provider.SAGEMAKER,
endpoint_config={
"endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
"region_name": os.getenv("AWS_REGION", "us-west-2"),
},
payload_config={"max_tokens": 256, "temperature": 0.2, "stream": False},
)Streaming uses the async stream() API (stream_text() is not available for SageMaker):
python
import asyncio
import os
from elsai_model.sagemaker import SageMakerAIModel
async def stream_response() -> None:
model = SageMakerAIModel(
endpoint_config={
"endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
"region_name": os.getenv("AWS_REGION", "us-west-2"),
},
payload_config={"max_tokens": 256, "temperature": 0.2, "stream": True},
)
messages = [{"role": "user", "content": [{"text": "Count from 1 to 5, one number per line."}]}]
async for event in model.stream(messages):
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"].get("delta", {})
if "text" in delta:
print(delta["text"], end="", flush=True)
print()
asyncio.run(stream_response())See also
- LLM Models overview — install, extras, factory
- LLM factory