NVIDIA

pip install -qU langchain-nvidia-ai-endpoints

import getpass
import os

if os.environ.get("NVIDIA_API_KEY", "").startswith("nvapi-"):
    print("环境中已存在有效的 NVIDIA_API_KEY。删除后可重置")
else:
    nvapi_key = getpass.getpass("NVAPI 密钥（以 nvapi- 开头）：")
    assert nvapi_key.startswith(
        "nvapi-"
    ), f"{nvapi_key[:5]}... 不是有效密钥"
    os.environ["NVIDIA_API_KEY"] = nvapi_key

from langchain_nvidia_ai_endpoints import ChatNVIDIA

# Nemotron 3 Super — 高效推理和智能体任务
llm = ChatNVIDIA(model="nvidia/nemotron-3-super-120b-a12b")
result = llm.invoke("为竞品分析规划一个三步研究工作流。")
print(result.content)

from langchain_nvidia_ai_endpoints import ChatNVIDIADynamo

llm = ChatNVIDIADynamo(
    base_url="http://localhost:8099/v1",
    model="nvidia/nemotron-3-super-120b-a12b",
    osl=512,             # 预期输出序列长度（token）
    iat=250,             # 预期到达间隔时间（毫秒）
    latency_sensitivity=1.0,
    priority=1,
)
result = llm.invoke("用一句话总结 KV 缓存路由。")
print(result.content)

from langchain_nvidia_ai_endpoints import NVIDIAEmbeddings

embedder = NVIDIAEmbeddings(model="NV-Embed-QA")
embedder.embed_query("今天的温度是多少？")

from langchain_core.documents import Document
from langchain_nvidia_ai_endpoints import NVIDIARerank

ranker = NVIDIARerank(model="nvidia/llama-3.2-nv-rerankqa-1b-v1")
docs = ranker.compress_documents(
    query="什么是 GPU 内存带宽？",
    documents=[Document(page_content=p) for p in passages],
)

from langchain_nvidia_ai_endpoints import NVIDIARAGRetriever

retriever = NVIDIARAGRetriever(base_url="http://localhost:8081", k=4)
docs = retriever.invoke("什么是 NVIDIA NIM？")

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")

pip install -qU langchain-nvidia-langgraph

from langchain_nvidia_langgraph.graph import StateGraph, OptimizationConfig
from langgraph.graph import END
from typing import TypedDict

class AgentState(TypedDict):
  ...

graph = StateGraph(AgentState)
app = graph.compile(optimization=OptimizationConfig(enable_parallel=True))

from langgraph.graph import StateGraph as LangGraphStateGraph
graph = LangGraphStateGraph(AgentState)
app = with_app_compile(graph).compile(optimization=OptimizationConfig(enable_parallel=True))

from langchain_nvidia_langgraph.graph import sequential, depends_on, speculation_unsafe

# 防止某个节点被并行化（例如，它会写入共享状态）
@sequential
def write_to_db(state):
    ...

# 声明一个未在图边中表达的依赖关系
@depends_on("write_to_db")
def next_action(state):
    ...

app = graph.compile(optimization=OptimizationConfig(enable_speculation=True))