Chatbot with Profile Schema¶
Review¶
We introduced the LangGraph Memory Store as a way to save and retrieve long-term memories.
We built a simple chatbot that uses both short-term (within-thread) and long-term (across-thread) memory.
It saved long-term semantic memory (facts about the user) "in the hot path", as the user is chatting with it.
Goals¶
Our chatbot saved memories as a string. In practice, we often want memories to have a structure.
For example, memories can be a single, continuously updated schema.
In our case, we want this to be a single user profile.
We'll extend our chatbot to save semantic memories to a single user profile.
We'll also introduce a library, Trustcall, to update this schema with new information.
In [ ]:
from dotenv import load_dotenv
load_dotenv()
In [7]:
%%capture --no-stderr
%pip install -U langchain-deepseek langgraph trustcall langchain_core
In [1]:
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "langchain-academy"
Defining a user profile schema¶
Python has many different types for structured data, such as TypedDict, Dictionaries, JSON, and Pydantic.
Let's start by using TypedDict to define a user profile schema.
In [3]:
from typing import TypedDict, List
class UserProfile(TypedDict):
"""User profile schema with typed fields"""
user_name: str # The user's preferred name
interests: List[str] # A list of the user's interests
Saving a schema to the store¶
The LangGraph Store accepts any Python dictionary as the value.
In [4]:
# TypedDict instance
user_profile: UserProfile = {
"user_name": "Lance",
"interests": ["biking", "technology", "coffee"]
}
user_profile
Out[4]:
{'user_name': 'Lance', 'interests': ['biking', 'technology', 'coffee']}
We use the put method to save the TypedDict to the store.
In [5]:
import uuid
from langgraph.store.memory import InMemoryStore
# Initialize the in-memory store
in_memory_store = InMemoryStore()
# Namespace for the memory to save
user_id = "1"
namespace_for_memory = (user_id, "memory")
# Save a memory to namespace as key and value
key = "user_profile"
value = user_profile
in_memory_store.put(namespace_for_memory, key, value)
We use search to retrieve objects from the store by namespace.
In [6]:
# Search
for m in in_memory_store.search(namespace_for_memory):
print(m.dict())
{'value': {'user_name': 'Lance', 'interests': ['biking', 'technology', 'coffee']}, 'key': 'user_profile', 'namespace': ['1', 'memory'], 'created_at': '2024-11-04T23:37:34.871675+00:00', 'updated_at': '2024-11-04T23:37:34.871680+00:00'}
We can also use get to retrieve a specific object by namespace and key.
In [7]:
# Get the memory by namespace and key
profile = in_memory_store.get(namespace_for_memory, "user_profile")
profile.value
Out[7]:
{'user_name': 'Lance', 'interests': ['biking', 'technology', 'coffee']}
Chatbot with profile schema¶
Now we know how to specify a schema for the memories and save it to the store.
Now, how do we actually create memories with this particular schema?
In our chatbot, we want to create memories from a user chat.
This is where the concept of structured outputs is useful.
LangChain's chat model interface has a with_structured_output method to enforce structured output.
This is useful when we want to enforce that the output conforms to a schema, and it parses the output for us.
Let's pass the UserProfile schema we created to the with_structured_output method.
We can then invoke the chat model with a list of messages and get a structured output that conforms to our schema.
In [8]:
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
# Initialize the model
model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)
# Bind schema to model
model_with_structure = model.with_structured_output(UserProfile)
# Invoke the model to produce structured output that matches the schema
structured_output = model_with_structure.invoke([HumanMessage("My name is Lance, I like to bike.")])
structured_output
Out[8]:
{'user_name': 'Lance', 'interests': ['biking']}
Now, let's use this with our chatbot.
This only requires minor changes to the write_memory function.
We use model_with_structure, as defined above, to produce a profile that matches our schema.
In [9]:
from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.store.base import BaseStore
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.runnables.config import RunnableConfig
# Chatbot instruction
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""
# Create new memory from the chat history and any existing memory
CREATE_MEMORY_INSTRUCTION = """Create or update a user profile memory based on the user's chat history.
This will be saved for long-term memory. If there is an existing memory, simply update it.
Here is the existing memory (it may be empty): {memory}"""
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Load memory from the store and use it to personalize the chatbot's response."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Retrieve memory from the store
namespace = ("memory", user_id)
existing_memory = store.get(namespace, "user_memory")
# Format the memories for the system prompt
if existing_memory and existing_memory.value:
memory_dict = existing_memory.value
formatted_memory = (
f"Name: {memory_dict.get('user_name', 'Unknown')}\n"
f"Interests: {', '.join(memory_dict.get('interests', []))}"
)
else:
formatted_memory = None
# Format the memory in the system prompt
system_msg = MODEL_SYSTEM_MESSAGE.format(memory=formatted_memory)
# Respond using memory as well as the chat history
response = model.invoke([SystemMessage(content=system_msg)]+state["messages"])
return {"messages": response}
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Reflect on the chat history and save a memory to the store."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Retrieve existing memory from the store
namespace = ("memory", user_id)
existing_memory = store.get(namespace, "user_memory")
# Format the memories for the system prompt
if existing_memory and existing_memory.value:
memory_dict = existing_memory.value
formatted_memory = (
f"Name: {memory_dict.get('user_name', 'Unknown')}\n"
f"Interests: {', '.join(memory_dict.get('interests', []))}"
)
else:
formatted_memory = None
# Format the existing memory in the instruction
system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=formatted_memory)
# Invoke the model to produce structured output that matches the schema
new_memory = model_with_structure.invoke([SystemMessage(content=system_msg)]+state['messages'])
# Overwrite the existing use profile memory
key = "user_memory"
store.put(namespace, key, new_memory)
# Define the graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)
# Store for long-term (across-thread) memory
across_thread_memory = InMemoryStore()
# Checkpointer for short-term (within-thread) memory
within_thread_memory = MemorySaver()
# Compile the graph with the checkpointer fir and store
graph = builder.compile(checkpointer=within_thread_memory, store=across_thread_memory)
# View
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))
In [10]:
# We supply a thread ID for short-term (within-thread) memory
# We supply a user ID for long-term (across-thread) memory
config = {"configurable": {"thread_id": "1", "user_id": "1"}}
# User input
input_messages = [HumanMessage(content="Hi, my name is Lance and I like to bike around San Francisco and eat at bakeries.")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
================================ Human Message ================================= Hi, my name is Lance and I like to bike around San Francisco and eat at bakeries. ================================== Ai Message ================================== Hi Lance! It's great to meet you. Biking around San Francisco sounds like a fantastic way to explore the city, and there are so many amazing bakeries to try. Do you have any favorite bakeries or biking routes in the city?
Let's check the memory in the store.
We can see that the memory is a dictionary that matches our schema.
In [11]:
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = across_thread_memory.get(namespace, "user_memory")
existing_memory.value
Out[11]:
{'user_name': 'Lance', 'interests': ['biking', 'bakeries', 'San Francisco']}
When can this fail?¶
with_structured_output is very useful, but what happens if we're working with a more complex schema?
Here's an example of a more complex schema, which we'll test below.
This is a Pydantic model that describes a user's preferences for communication and trust fall.
In [18]:
from typing import List, Optional
class OutputFormat(BaseModel):
preference: str
sentence_preference_revealed: str
class TelegramPreferences(BaseModel):
preferred_encoding: Optional[List[OutputFormat]] = None
favorite_telegram_operators: Optional[List[OutputFormat]] = None
preferred_telegram_paper: Optional[List[OutputFormat]] = None
class MorseCode(BaseModel):
preferred_key_type: Optional[List[OutputFormat]] = None
favorite_morse_abbreviations: Optional[List[OutputFormat]] = None
class Semaphore(BaseModel):
preferred_flag_color: Optional[List[OutputFormat]] = None
semaphore_skill_level: Optional[List[OutputFormat]] = None
class TrustFallPreferences(BaseModel):
preferred_fall_height: Optional[List[OutputFormat]] = None
trust_level: Optional[List[OutputFormat]] = None
preferred_catching_technique: Optional[List[OutputFormat]] = None
class CommunicationPreferences(BaseModel):
telegram: TelegramPreferences
morse_code: MorseCode
semaphore: Semaphore
class UserPreferences(BaseModel):
communication_preferences: CommunicationPreferences
trust_fall_preferences: TrustFallPreferences
class TelegramAndTrustFallPreferences(BaseModel):
pertinent_user_preferences: UserPreferences
Now, let's try extraction of this schema using the with_structured_output method.
In [19]:
from pydantic import ValidationError
# Bind schema to model
model_with_structure = model.with_structured_output(TelegramAndTrustFallPreferences)
# Conversation
conversation = """Operator: How may I assist with your telegram, sir?
Customer: I need to send a message about our trust fall exercise.
Operator: Certainly. Morse code or standard encoding?
Customer: Morse, please. I love using a straight key.
Operator: Excellent. What's your message?
Customer: Tell him I'm ready for a higher fall, and I prefer the diamond formation for catching.
Operator: Done. Shall I use our "Daredevil" paper for this daring message?
Customer: Perfect! Send it by your fastest carrier pigeon.
Operator: It'll be there within the hour, sir."""
# Invoke the model
try:
model_with_structure.invoke(f"""Extract the preferences from the following conversation:
<convo>
{conversation}
</convo>""")
except ValidationError as e:
print(e)
1 validation error for TelegramAndTrustFallPreferences
pertinent_user_preferences.communication_preferences.semaphore
Input should be a valid dictionary or instance of Semaphore [type=model_type, input_value=None, input_type=NoneType]
For further information visit https://errors.pydantic.dev/2.9/v/model_type
If we naively extract more complex schemas, even using high capacity model like deepseek-v4-pro, it is prone to failure.
Trustcall for creating and updating profile schemas¶
As we can see, working with schemas can be tricky.
Complex schemas can be difficult to extract.
In addition, updating even simple schemas can pose challenges.
Consider our above chatbot.
We regenerated the profile schema from scratch each time we chose to save a new memory.
This is inefficient, potentially wasting model tokens if the schema contains a lot of information to re-generate each time.
Worse, we may loose information when regenerating the profile from scratch.
Addressing these problems is the motivation for TrustCall!
This is an open-source library for updating JSON schemas developed by one Will Fu-Hinthorn on the LangChain team.
It's motivated by exactly these challenges while working on memory.
Let's first show simple usage of extraction with TrustCall on this list of messages.
In [20]:
# Conversation
conversation = [HumanMessage(content="Hi, I'm Lance."),
AIMessage(content="Nice to meet you, Lance."),
HumanMessage(content="I really like biking around San Francisco.")]
We use create_extractor, passing in the model as well as our schema as a tool.
With TrustCall, can supply supply the schema in various ways.
For example, we can pass a JSON object / Python dictionary or Pydantic model.
Under the hood, TrustCall uses tool calling to produce structured output from an input list of messages.
To force Trustcall to produce structured output, we can include the schema name in the tool_choice argument.
We can invoke the extractor with the above conversation.
In [21]:
from trustcall import create_extractor
# Schema
class UserProfile(BaseModel):
"""User profile schema with typed fields"""
user_name: str = Field(description="The user's preferred name")
interests: List[str] = Field(description="A list of the user's interests")
# Initialize the model
model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)
# Create the extractor
trustcall_extractor = create_extractor(
model,
tools=[UserProfile],
tool_choice="UserProfile"
)
# Instruction
system_msg = "Extract the user profile from the following conversation"
# Invoke the extractor
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=system_msg)]+conversation})
When we invoke the extractor, we get a few things:
messages: The list ofAIMessagesthat contain the tool calls.responses: The resulting parsed tool calls that match our schema.response_metadata: Applicable if updating existing tool calls. It says which of the responses correspond to which of the existing objects.
In [17]:
for m in result["messages"]:
m.pretty_print()
================================== Ai Message ==================================
Tool Calls:
UserProfile (call_spGGUsoaUFXU7oOrUNCASzfL)
Call ID: call_spGGUsoaUFXU7oOrUNCASzfL
Args:
user_name: Lance
interests: ['biking around San Francisco']
In [18]:
schema = result["responses"]
schema
Out[18]:
[UserProfile(user_name='Lance', interests=['biking around San Francisco'])]
In [19]:
schema[0].model_dump()
Out[19]:
{'user_name': 'Lance', 'interests': ['biking around San Francisco']}
In [20]:
result["response_metadata"]
Out[20]:
[{'id': 'call_spGGUsoaUFXU7oOrUNCASzfL'}]
Let's see how we can use it to update the profile.
For updating, TrustCall takes a set of messages as well as the existing schema.
The central idea is that it prompts the model to produce a JSON Patch to update only the relevant parts of the schema.
This is less error-prone than naively overwriting the entire schema.
It's also more efficient since the model only needs to generate the parts of the schema that have changed.
We can save the existing schema as a dict.
We can use model_dump() to serialize a Pydantic model instance into a dict.
We pass it to the "existing" argument along with the schema name, UserProfile.
In [21]:
# Update the conversation
updated_conversation = [HumanMessage(content="Hi, I'm Lance."),
AIMessage(content="Nice to meet you, Lance."),
HumanMessage(content="I really like biking around San Francisco."),
AIMessage(content="San Francisco is a great city! Where do you go after biking?"),
HumanMessage(content="I really like to go to a bakery after biking."),]
# Update the instruction
system_msg = f"""Update the memory (JSON doc) to incorporate new information from the following conversation"""
# Invoke the extractor with the updated instruction and existing profile with the corresponding tool name (UserProfile)
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=system_msg)]+updated_conversation},
{"existing": {"UserProfile": schema[0].model_dump()}})
In [22]:
for m in result["messages"]:
m.pretty_print()
================================== Ai Message ==================================
Tool Calls:
UserProfile (call_WeZl0ACfQStxblim0ps8LNKT)
Call ID: call_WeZl0ACfQStxblim0ps8LNKT
Args:
user_name: Lance
interests: ['biking', 'visiting bakeries']
In [23]:
result["response_metadata"]
Out[23]:
[{'id': 'call_WeZl0ACfQStxblim0ps8LNKT'}]
In [24]:
updated_schema = result["responses"][0]
updated_schema.model_dump()
Out[24]:
{'user_name': 'Lance', 'interests': ['biking', 'visiting bakeries']}
LangSmith trace:
https://smith.langchain.com/public/229eae22-1edb-44c6-93e6-489124a43968/r
Now, let's also test Trustcall on the challenging schema that we saw earlier.
In [25]:
bound = create_extractor(
model,
tools=[TelegramAndTrustFallPreferences],
tool_choice="TelegramAndTrustFallPreferences",
)
# Conversation
conversation = """Operator: How may I assist with your telegram, sir?
Customer: I need to send a message about our trust fall exercise.
Operator: Certainly. Morse code or standard encoding?
Customer: Morse, please. I love using a straight key.
Operator: Excellent. What's your message?
Customer: Tell him I'm ready for a higher fall, and I prefer the diamond formation for catching.
Operator: Done. Shall I use our "Daredevil" paper for this daring message?
Customer: Perfect! Send it by your fastest carrier pigeon.
Operator: It'll be there within the hour, sir."""
result = bound.invoke(
f"""Extract the preferences from the following conversation:
<convo>
{conversation}
</convo>"""
)
# Extract the preferences
result["responses"][0]
Out[25]:
TelegramAndTrustFallPreferences(pertinent_user_preferences=UserPreferences(communication_preferences=CommunicationPreferences(telegram=TelegramPreferences(preferred_encoding=[OutputFormat(preference='standard encoding', sentence_preference_revealed='standard encoding')], favorite_telegram_operators=None, preferred_telegram_paper=[OutputFormat(preference='Daredevil', sentence_preference_revealed='Daredevil')]), morse_code=MorseCode(preferred_key_type=[OutputFormat(preference='straight key', sentence_preference_revealed='straight key')], favorite_morse_abbreviations=None), semaphore=Semaphore(preferred_flag_color=None, semaphore_skill_level=None)), trust_fall_preferences=TrustFallPreferences(preferred_fall_height=[OutputFormat(preference='higher', sentence_preference_revealed='higher')], trust_level=None, preferred_catching_technique=[OutputFormat(preference='diamond formation', sentence_preference_revealed='diamond formation')])))
Trace:
https://smith.langchain.com/public/5cd23009-3e05-4b00-99f0-c66ee3edd06e/r
For more examples, you can see an overview video here.
Chatbot with profile schema updating¶
Now, let's bring Trustcall into our chatbot to create and update a memory profile.
In [22]:
from IPython.display import Image, display
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.runnables.config import RunnableConfig
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore
# Initialize the model
model = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)
# Schema
class UserProfile(BaseModel):
""" Profile of a user """
user_name: str = Field(description="The user's preferred name")
user_location: str = Field(description="The user's location")
interests: list = Field(description="A list of the user's interests")
# Create the extractor
trustcall_extractor = create_extractor(
model,
tools=[UserProfile],
tool_choice="UserProfile", # Enforces use of the UserProfile tool
)
# Chatbot instruction
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""
# Extraction instruction
TRUSTCALL_INSTRUCTION = """Create or update the memory (JSON doc) to incorporate information from the following conversation:"""
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Load memory from the store and use it to personalize the chatbot's response."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Retrieve memory from the store
namespace = ("memory", user_id)
existing_memory = store.get(namespace, "user_memory")
# Format the memories for the system prompt
if existing_memory and existing_memory.value:
memory_dict = existing_memory.value
formatted_memory = (
f"Name: {memory_dict.get('user_name', 'Unknown')}\n"
f"Location: {memory_dict.get('user_location', 'Unknown')}\n"
f"Interests: {', '.join(memory_dict.get('interests', []))}"
)
else:
formatted_memory = None
# Format the memory in the system prompt
system_msg = MODEL_SYSTEM_MESSAGE.format(memory=formatted_memory)
# Respond using memory as well as the chat history
response = model.invoke([SystemMessage(content=system_msg)]+state["messages"])
return {"messages": response}
def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
"""Reflect on the chat history and save a memory to the store."""
# Get the user ID from the config
user_id = config["configurable"]["user_id"]
# Retrieve existing memory from the store
namespace = ("memory", user_id)
existing_memory = store.get(namespace, "user_memory")
# Get the profile as the value from the list, and convert it to a JSON doc
existing_profile = {"UserProfile": existing_memory.value} if existing_memory else None
# Invoke the extractor
result = trustcall_extractor.invoke({"messages": [SystemMessage(content=TRUSTCALL_INSTRUCTION)]+state["messages"], "existing": existing_profile})
# Get the updated profile as a JSON object
updated_profile = result["responses"][0].model_dump()
# Save the updated profile
key = "user_memory"
store.put(namespace, key, updated_profile)
# Define the graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)
# Store for long-term (across-thread) memory
across_thread_memory = InMemoryStore()
# Checkpointer for short-term (within-thread) memory
within_thread_memory = MemorySaver()
# Compile the graph with the checkpointer fir and store
graph = builder.compile(checkpointer=within_thread_memory, store=across_thread_memory)
# View
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))
In [23]:
# We supply a thread ID for short-term (within-thread) memory
# We supply a user ID for long-term (across-thread) memory
config = {"configurable": {"thread_id": "1", "user_id": "1"}}
# User input
input_messages = [HumanMessage(content="Hi, my name is Lance")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
================================ Human Message ================================= Hi, my name is Lance ================================== Ai Message ================================== Hello, Lance! It's nice to meet you. How can I assist you today?
In [24]:
# User input
input_messages = [HumanMessage(content="I like to bike around San Francisco")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
================================ Human Message ================================= I like to bike around San Francisco ================================== Ai Message ================================== That sounds like a great way to explore the city! San Francisco has some beautiful routes and views. Do you have any favorite trails or spots you like to visit while biking?
In [25]:
# Namespace for the memory to save
user_id = "1"
namespace = ("memory", user_id)
existing_memory = across_thread_memory.get(namespace, "user_memory")
existing_memory.dict()
Out[25]:
{'value': {'user_name': 'Lance',
'user_location': 'San Francisco',
'interests': ['biking']},
'key': 'user_memory',
'namespace': ['memory', '1'],
'created_at': '2024-11-04T23:51:17.662428+00:00',
'updated_at': '2024-11-04T23:51:41.697652+00:00'}
In [26]:
# The user profile saved as a JSON object
existing_memory.value
Out[26]:
{'user_name': 'Lance',
'user_location': 'San Francisco',
'interests': ['biking']}
In [27]:
# User input
input_messages = [HumanMessage(content="I also enjoy going to bakeries")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
================================ Human Message ================================= I also enjoy going to bakeries ================================== Ai Message ================================== Biking and visiting bakeries sounds like a delightful combination! San Francisco has some fantastic bakeries. Do you have any favorites, or are you looking for new recommendations to try out?
Continue the conversation in a new thread.
In [28]:
# We supply a thread ID for short-term (within-thread) memory
# We supply a user ID for long-term (across-thread) memory
config = {"configurable": {"thread_id": "2", "user_id": "1"}}
# User input
input_messages = [HumanMessage(content="What bakeries do you recommend for me?")]
# Run the graph
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
================================ Human Message ================================= What bakeries do you recommend for me? ================================== Ai Message ================================== Since you're in San Francisco and enjoy going to bakeries, here are a few recommendations you might like: 1. **Tartine Bakery** - Known for its delicious bread and pastries, it's a must-visit for any bakery enthusiast. 2. **B. Patisserie** - Offers a delightful selection of French pastries, including their famous kouign-amann. 3. **Arsicault Bakery** - Renowned for its croissants, which have been praised as some of the best in the country. 4. **Craftsman and Wolves** - Known for their inventive pastries and the "Rebel Within," a savory muffin with a soft-cooked egg inside. 5. **Mr. Holmes Bakehouse** - Famous for their cruffins and other creative pastries. These spots should offer a great variety of treats for you to enjoy. Happy bakery hopping!
In [ ]: