ChatNVIDIA 集成

pip install -qU langchain-nvidia-ai-endpoints

import getpass
import os

if os.environ.get("NVIDIA_API_KEY", "").startswith("nvapi-"):
    print("Valid NVIDIA_API_KEY already in environment. Delete to reset")
else:
    nvapi_key = getpass.getpass("NVAPI Key (starts with nvapi-): ")
    assert nvapi_key.startswith(
        "nvapi-"
    ), f"{nvapi_key[:5]}... is not a valid key"
    os.environ["NVIDIA_API_KEY"] = nvapi_key

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

from langchain_nvidia_ai_endpoints import ChatNVIDIA

# Nemotron 3 Nano — 高效推理和智能体任务
llm = ChatNVIDIA(model="nvidia/nemotron-3-super-120b-a12b")

llm = ChatNVIDIA(model="mistralai/mixtral-8x7b-instruct-v0.1")

result = llm.invoke("Write a ballad about LangChain.")
print(result.content)

from langchain_nvidia_ai_endpoints import ChatNVIDIA, NVIDIAEmbeddings, NVIDIARerank

# 连接到在 localhost:8000 运行的聊天 NIM，指定模型
llm = ChatNVIDIA(base_url="http://localhost:8000/v1", model="nvidia/nemotron-3-super-120b-a12b")

# 连接到在 localhost:8080 运行的嵌入 NIM
embedder = NVIDIAEmbeddings(base_url="http://localhost:8080/v1")

# 连接到在 localhost:2016 运行的重排序 NIM
ranker = NVIDIARerank(base_url="http://localhost:2016/v1")

print(llm.batch(["What's 2*3?", "What's 2*6?"]))
# 或通过异步 API
# await llm.abatch(["What's 2*3?", "What's 2*6?"])

for chunk in llm.stream("How far can a seagull fly in one day?"):
    # 显示 token 分隔
    print(chunk.content, end="|")

async for chunk in llm.astream(
    "How long does it take for monarch butterflies to migrate?"
):
    print(chunk.content, end="|")

ChatNVIDIA.get_available_models()
# llm.get_available_models()

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

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful research assistant. Think step by step."),
        ("user", "{input}"),
    ]
)
chain = prompt | ChatNVIDIA(model="nvidia/nemotron-3-super-120b-a12b") | StrOutputParser()

for txt in chain.stream({"input": "What are the key considerations when designing a multi-agent RAG system?"}):
    print(txt, end="")

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

prompt = ChatPromptTemplate.from_messages(
    [("system", "You are a helpful AI assistant named Fred."), ("user", "{input}")]
)
chain = prompt | ChatNVIDIA(model="nvidia/nemotron-3-super-120b-a12b") | StrOutputParser()

for txt in chain.stream({"input": "What's your name?"}):
    print(txt, end="")

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are an expert coding AI. Respond only in valid python; no narration whatsoever.",
        ),
        ("user", "{input}"),
    ]
)
chain = prompt | ChatNVIDIA(model="meta/codellama-70b") | StrOutputParser()

for txt in chain.stream({"input": "How do I solve this fizz buzz problem?"}):
    print(txt, end="")

import IPython
import requests

image_url = "https://www.nvidia.com/content/dam/en-zz/Solutions/research/ai-playground/nvidia-picasso-3c33-p@2x.jpg"  ## 大图像
image_content = requests.get(image_url).content

IPython.display.Image(image_content)

from langchain_nvidia_ai_endpoints import ChatNVIDIA

llm = ChatNVIDIA(model="nvidia/neva-22b")

from langchain.messages import HumanMessage

llm.invoke(
    [
        HumanMessage(
            content=[
                {"type": "text", "text": "Describe this image:"},
                {"type": "image_url", "image_url": {"url": image_url}},
            ]
        )
    ]
)

import IPython
import requests

image_url = "https://picsum.photos/seed/kitten/300/200"
image_content = requests.get(image_url).content

IPython.display.Image(image_content)

import base64

from langchain.messages import HumanMessage

## 适用于较简单的图像。对于较大的图像，请参阅实际实现
b64_string = base64.b64encode(image_content).decode("utf-8")

llm.invoke(
    [
        HumanMessage(
            content=[
                {"type": "text", "text": "Describe this image:"},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/png;base64,{b64_string}"},
                },
            ]
        )
    ]
)

base64_with_mime_type = f"data:image/png;base64,{b64_string}"
llm.invoke(f'What\'s in this image?\n<img src="{base64_with_mime_type}" />')

pip install -qU langchain

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# store 是一个将会话 ID 映射到其相应聊天历史记录的字典。
store = {}  # 内存维护在链外部


# 返回给定会话 ID 的聊天历史记录的函数。
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]


chat = ChatNVIDIA(
    model="nvidia/nemotron-3-super-120b-a12b",
    temperature=0.1,
    max_tokens=100,
    top_p=1.0,
)

# 定义一个 RunnableConfig 对象，包含一个 `configurable` 键。session_id 决定线程
config = {"configurable": {"session_id": "1"}}

conversation = RunnableWithMessageHistory(
    chat,
    get_session_history,
)

conversation.invoke(
    "Hi I'm Srijan Dubey.",  # 输入或查询
    config=config,
)

conversation.invoke(
    "I'm doing well! Just having a conversation with an AI.",
    config=config,
)

conversation.invoke(
    "Tell me about yourself.",
    config=config,
)

tool_models = [
    model for model in ChatNVIDIA.get_available_models() if model.supports_tools
]
tool_models

from langchain.tools import tool
from pydantic import Field


@tool
def get_current_weather(
    location: str = Field(description="The location to get the weather for."),
):
    """Get the current weather for a location."""
    ...


llm = ChatNVIDIA(model=tool_models[0].id).bind_tools(tools=[get_current_weather])
response = llm.invoke("What is the weather in Boston?")
response.tool_calls

from langchain_nvidia_ai_endpoints import ChatNVIDIA, ChatNVIDIADynamo

BASE_URL = "http://localhost:8099/v1"
MODEL = "your-model-name"

# 标准 ChatNVIDIA — 无 Dynamo 提示
llm_standard = ChatNVIDIA(base_url=BASE_URL, model=MODEL)

# ChatNVIDIADynamo — 相同的接口，自动注入 agent_hints
llm = ChatNVIDIADynamo(base_url=BASE_URL, model=MODEL)

result = llm.invoke("What is KV cache optimization?")
print(result.content)

# 高优先级：短响应，延迟关键
llm_critical = ChatNVIDIADynamo(
    base_url=BASE_URL,
    model=MODEL,
    osl=20,
    priority=0,
    latency_sensitivity=10.0,
)

# 低优先级：长响应，延迟容忍
llm_background = ChatNVIDIADynamo(
    base_url=BASE_URL,
    model=MODEL,
    osl=512,
    priority=10,
    latency_sensitivity=0,
)

result = llm.invoke(
    "Classify this as positive or negative: 'I love this product!'",
    osl=10,
    iat=100,
    latency_sensitivity=1.0,
    priority=10,
)
print(result.content)

for chunk in llm_critical.stream("Give a one-sentence summary of GPU computing."):
    print(chunk.content, end="", flush=True)

import json

payload = llm_critical._get_payload(
    inputs=[{"role": "user", "content": "Hello!"}],
    stop=None,
)
print(json.dumps(payload["nvext"], indent=2))

{
  "agent_hints": {
    "prefix_id": "langchain-dynamo-a1b2c3d4e5f6",
    "osl": 20,
    "iat": 250,
    "latency_sensitivity": 1.0,
    "priority": 10
  }
}