ChatAmazonNova 集成

import getpass
import os

if "NOVA_API_KEY" not in os.environ:
    os.environ["NOVA_API_KEY"] = getpass.getpass("Enter your Nova API key: ")

if "NOVA_BASE_URL" not in os.environ:
    os.environ["NOVA_BASE_URL"] = getpass.getpass("Enter your Nova base URL: ")

os.environ["LANGSMITH_API_KEY"] = getpass.getpass("Enter your LangSmith API key: ")
os.environ["LANGSMITH_TRACING"] = "true"

from langchain_amazon_nova import ChatAmazonNova

model = ChatAmazonNova(
    model="nova-2-lite-v1",
    temperature=0.7,
    max_tokens=2048,
    timeout=None,
    max_retries=2,
    # other params...
)

messages = [
    (
        "system",
        "You are a helpful assistant that translates English to French. Translate the user sentence.",
    ),
    ("user", "I love programming."),
]
ai_msg = model.invoke(messages)
ai_msg

AIMessage(content="J'adore la programmation.", response_metadata={'model': 'nova-2-lite-v1', 'finish_reason': 'stop'}, id='run-12345678-1234-1234-1234-123456789abc', usage_metadata={'input_tokens': 29, 'output_tokens': 8, 'total_tokens': 37})

print(ai_msg.content)

J'adore la programmation.

ai_msg.content_blocks

for chunk in model.stream(messages):
    print(chunk.content, end="", flush=True)

J'adore la programmation.

import asyncio

async def main():
    async for chunk in model.astream(messages):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

from pydantic import BaseModel, Field

class GetWeather(BaseModel):
    """Get the weather for a location."""

    location: str = Field(description="City name")

model_with_tools = model.bind_tools([GetWeather])
response = model_with_tools.invoke("What's the weather in Paris?")
print(response.tool_calls)

[{'name': 'GetWeather', 'args': {'location': 'Paris'}, 'id': 'call_abc123', 'type': 'tool_call'}]

response.tool_calls

[{'name': 'GetWeather',
  'args': {'location': 'Paris'},
  'id': 'call_abc123',
  'type': 'tool_call'}]

from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """Get the weather for a location."""
    return f"Weather in {location}: Sunny, 72°F"

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("What's the weather in San Francisco?")

# 模型决定是否调用工具（默认）
model_auto = model.bind_tools([get_weather], tool_choice="auto")

# 模型必须调用一个工具
model_required = model.bind_tools([get_weather], tool_choice="required")

# 模型不能调用工具
model_none = model.bind_tools([get_weather], tool_choice="none")

from langchain_amazon_nova import ChatAmazonNova

model_with_grounding = ChatAmazonNova(
    model="nova-2-lite-v1",
    system_tools=["nova_grounding"],
)

response = model_with_grounding.invoke("What are the latest developments in AI?")

from langchain_amazon_nova import ChatAmazonNova

model_with_code = ChatAmazonNova(
    model="nova-2-lite-v1",
    system_tools=["nova_code_interpreter"],
)

response = model_with_code.invoke("Calculate the fibonacci sequence up to the 10th number")

from langchain_amazon_nova import ChatAmazonNova

model_with_tools = ChatAmazonNova(
    model="nova-2-lite-v1",
    system_tools=["nova_grounding", "nova_code_interpreter"],
)

response = model_with_tools.invoke(
    "Search for the current price of Bitcoin and calculate its 7-day moving average"
)

from langchain_amazon_nova import ChatAmazonNova

model = ChatAmazonNova(model="nova-2-lite-v1")

# 为此特定调用启用 grounding
response = model.invoke(
    "What's the weather today?",
    system_tools=["nova_grounding"]
)

from pydantic import BaseModel, Field
from langchain_amazon_nova import ChatAmazonNova

class Person(BaseModel):
    """Information about a person."""
    name: str = Field(description="The person's name")
    age: int = Field(description="The person's age")

model = ChatAmazonNova(model="nova-pro-v1")
structured_model = model.with_structured_output(Person)

result = structured_model.invoke("John is 30 years old")
print(result)

Person(name='John', age=30)

json_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"}
    },
    "required": ["name", "age"]
}

structured_model = model.with_structured_output(json_schema)
result = structured_model.invoke("Sarah is 28 years old")
print(result)

{'name': 'Sarah', 'age': 28}

from pydantic import BaseModel, Field

class Person(BaseModel):
    """Information about a person."""
    name: str = Field(description="The person's name")
    age: int = Field(description="The person's age")

structured_model = model.with_structured_output(Person)

for chunk in structured_model.stream("Michael is 35 years old"):
    print(chunk)

Person(name='Michael', age=35)

structured_model = model.with_structured_output(Person, include_raw=True)
result = structured_model.invoke("John is 30 years old")

print(f"Parsed: {result['parsed']}")
print(f"Raw message: {result['raw']}")

Parsed: Person(name='John', age=30)
Raw message: AIMessage(content='', additional_kwargs={'tool_calls': [...]}, ...)

from pydantic import BaseModel, Field
from typing import List

class Address(BaseModel):
    """A physical address."""
    street: str
    city: str
    country: str

class Person(BaseModel):
    """Information about a person."""
    name: str
    age: int
    addresses: List[Address] = Field(description="List of addresses")

structured_model = model.with_structured_output(Person)
result = structured_model.invoke(
    "John is 30 years old. He lives at 123 Main St in Seattle, USA "
    "and has a vacation home at 456 Beach Rd in Miami, USA."
)
print(result)

import asyncio

async def main():
    messages = [
        ("system", "You are a helpful assistant."),
        ("human", "What is the capital of France?"),
    ]
    response = await model.ainvoke(messages)
    print(response.content)

asyncio.run(main())

The capital of France is Paris.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that translates {input_language} to {output_language}."),
    ("human", "{text}"),
])

chain = prompt | model | StrOutputParser()

result = chain.invoke({
    "input_language": "English",
    "output_language": "Spanish",
    "text": "Hello, how are you?"
})
print(result)

Hola, ¿cómo estás?

model = ChatAmazonNova(
    model="nova-2-lite-v1",
    max_retries=3,  # Number of retries on failure
    timeout=60.0,   # Request timeout in seconds
)

model_with_custom_retry = model.with_retry(
    stop_after_attempt=5,
    wait_exponential_jitter=True,
)

import os
print(f"API Key set: {'NOVA_API_KEY' in os.environ}")
print(f"Base URL: {os.environ.get('NOVA_BASE_URL', 'Not set')}")

# 客户端已自动配置为不压缩
model = ChatAmazonNova(model="nova-2-lite-v1")
# model.client 是已配置的 OpenAI 客户端