from langchain_aws import ChatBedrockConverse
from langchain_aws.middleware.prompt_caching import BedrockPromptCachingMiddleware
from langchain.agents import create_agent
from langchain_core.runnables import RunnableConfig
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import MemorySaver
@tool
def get_weather(city: str) -> str:
"""获取城市的当前天气。"""
return f"The weather in {city} is sunny and 72F."
# 系统提示必须超过 1,024 个 token 才能使缓存生效
LONG_PROMPT = (
"You are a helpful weather assistant with deep expertise in meteorology, "
"climate science, and atmospheric phenomena. When answering questions about "
"weather, provide accurate and up-to-date information. "
+ "You should always strive to give the most helpful response possible. " * 85
)
agent = create_agent(
model=ChatBedrockConverse(model="us.anthropic.claude-sonnet-4-5-20250929-v1:0"),
system_prompt=LONG_PROMPT,
tools=[get_weather],
middleware=[BedrockPromptCachingMiddleware(ttl="5m")],
checkpointer=MemorySaver(), # 持久化对话历史
)
# 使用 thread_id 来维护对话状态
config: RunnableConfig = {"configurable": {"thread_id": "user-123"}}
# 第一次调用：使用系统提示、工具和用户消息创建缓存
response = agent.invoke(
{"messages": [HumanMessage("What is the weather in Miami?")]}, config=config
)
last_msg = response["messages"][-1]
print(last_msg.content)
# 检查缓存 token 使用情况
um = last_msg.usage_metadata
if um:
details = um.get("input_token_details", {})
cache_read = details.get("cache_read", 0) or 0
cache_write = details.get("cache_creation", 0) or 0
print(f"Cache read: {cache_read}, Cache write: {cache_write}")
# 第二次调用：重用缓存的系统提示、工具和之前的消息
response = agent.invoke(
{"messages": [HumanMessage("How about Seattle?")]}, config=config
)
print(response["messages"][-1].content)
