Using Claude prompt caching with ChatPromptTemplate #25610
Replies: 2 comments 10 replies
-
|
Hey there, @felixwoestmann! I'm here to help you with your bug, answer any questions you have, and guide you on becoming a contributor. Let's dive in and work on resolving your issue together. To implement Claude's prompt caching with Here's how you can set it up: from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=cached_prompt, additional_kwargs={"cache-control": {"type": "ephemeral"}}),
HumanMessage(content=f"Here is the comment to analyze: {comment}")
])This ensures that the Additionally, make sure that your caching mechanism is properly configured within the def _generate_with_cache(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
if isinstance(self.cache, BaseCache):
llm_cache = self.cache
else:
llm_cache = get_llm_cache()
check_cache = self.cache or self.cache is None
if check_cache:
if llm_cache:
llm_string = self._get_llm_string(stop=stop, **kwargs)
prompt = dumps(messages)
cache_val = llm_cache.lookup(prompt, llm_string)
if isinstance(cache_val, list):
return ChatResult(generations=cache_val)
elif self.cache is None:
pass
else:
raise ValueError(
"Asked to cache, but no cache found at `langchain.cache`."
)
result = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
for idx, generation in enumerate(result.generations):
if run_manager and generation.message.id is None:
generation.message.id = f"run-{run_manager.run_id}-{idx}"
generation.message.response_metadata = _gen_info_and_msg_metadata(generation)
if len(result.generations) == 1 and result.llm_output is not None:
result.generations[0].message.response_metadata = {
**result.llm_output,
**result.generations[0].message.response_metadata,
}
if check_cache and llm_cache:
llm_cache.update(prompt, llm_string, result.generations)
return resultThis code ensures that the caching mechanism checks the cache for existing results and updates the cache with new results as needed [1]. By setting the |
Beta Was this translation helpful? Give feedback.
-
|
This is plain wrong. I want to use Claudes prompt caching, not the langchain internal mechanism. |
Beta Was this translation helpful? Give feedback.
-
|
To implement Claude's prompt caching using
Here's a sample implementation: import hashlib
import json
from typing import List, Optional, Any
class CustomCache:
def __init__(self):
self.cache = {}
def _hash_prompt(self, prompt: str) -> str:
return hashlib.sha256(prompt.encode()).hexdigest()
def lookup(self, prompt: str, llm_string: str) -> Optional[List[Any]]:
key = self._hash_prompt(prompt + llm_string)
return self.cache.get(key)
def update(self, prompt: str, llm_string: str, generations: List[Any]):
key = self._hash_prompt(prompt + llm_string)
self.cache[key] = generations
class CustomChatModel(BaseChatModel):
def __init__(self, cache: CustomCache):
self.cache = cache
def _generate_with_cache(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional<CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
llm_string = self._get_llm_string(stop=stop, **kwargs)
prompt = json.dumps([message.to_dict() for message in messages])
# Check custom cache
cache_val = self.cache.lookup(prompt, llm_string)
if cache_val:
return ChatResult(generations=cache_val)
# Generate result if not in cache
result = self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
# Update custom cache
self.cache.update(prompt, llm_string, result.generations)
return result
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional<CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> ChatResult:
# Implement the actual generation logic here
pass
# Usage
custom_cache = CustomCache()
chat_model = CustomChatModel(cache=custom_cache)In this implementation:
Additionally, you can use from langchain_core.prompts.chat import ChatPromptTemplate
refine_template = (
"We have the opportunity to refine the existing answer "
"(only if needed) with some more context below.\n"
"------------\n"
"{context_str}\n"
"------------\n"
"Given the new context, refine the original answer to better "
"answer the question. "
"If the context isn't useful, return the original answer."
)
CHAT_REFINE_PROMPT = ChatPromptTemplate.from_messages(
[("human", "{question}"), ("ai", "{existing_answer}"), ("human", refine_template)]
)You can also use other prompt templates like from langchain_core.prompts.prompt import PromptTemplate
_template = """Given the following conversation and a follow up question, rephrase the follow up question to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone question:"""
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(_template)
prompt_template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
{context}
Question: {question}
Helpful Answer:"""
QA_PROMPT = PromptTemplate(
template=prompt_template, input_variables=["context", "question"]
)This way, you can implement prompt caching without relying on LangChain's internal caching mechanism while leveraging |
Beta Was this translation helpful? Give feedback.
-
|
Is it possible to cache the multiple users? |
Beta Was this translation helpful? Give feedback.
-
|
Solved it by doing this: |
Beta Was this translation helpful? Give feedback.
-
|
This doesn't seem to do anything for me. Not getting any indication that the prompted has been cached. This really shouldn't be this difficult... |
Beta Was this translation helpful? Give feedback.
-
|
Is the cached part at least 1024 tokens? And did you check if tokens were cached by logging the response headers? |
Beta Was this translation helpful? Give feedback.
-
|
It seems that my problem was, I was using streaming (astream_events) instead of invoke. It was caching it but the events in the response from streaming don't show that. I only managed to verify it's working by looking at my billing in Anthropic, and seeing the "Cache Hit" billing go up every time I run. |
Beta Was this translation helpful? Give feedback.
-
Did you use f-string to combine the system instructions with the long(fixed) context to make a system prompt? |
Beta Was this translation helpful? Give feedback.
-
|
I could implement prompt caching using ChatPromptTemplate. I have a different concern. I have the exact same prompt which I used as a PromptTemplate and it worked as intended. Now, the performance has worsened when I use ChatPromptTemplate. Any comments or solution for this? |
Beta Was this translation helpful? Give feedback.
-
Checked other resources
Commit to Help
Example Code
Description
Hello,
I am currently trying to implement Claude's prompt caching in my existing application.
Setting the extra header to enable this Beta works well and the API is returning an overview of how many tokens were written to and read from the cache. However these are always 0, due to my setting the
cache-controlproperty of the message wrongly.On the X.com account of Langchain I found an example to set the
cache-controlproperty when providing messages as a Json.However I am using ChatPromptTemplates and was wondering how I would do it there, currently I am trying something like this:
System Info
System Information
Package Information
Optional packages not installed
Other Dependencies
Beta Was this translation helpful? Give feedback.
All reactions