Closed ahuang11 closed 1 week ago
I looked up all llm.invoke
and I think I want the prompts to be structured like:
prompts/
- assistant:
- check_validity.jinja2
- handle_exception.jinja2
- source_agent.jinja2
- chat_agent.jinja2
- chat_details_agent.jinja2
- table_list_agent.jinja2
- sql_agent.jinja2
- hvplot_agent.jinja2
- analysis_agent.jinja2
- vega_lite_agent.jinja2
- source_agent:
- n/a
- chat_agent:
- require_data.jinja2
- system.jinja2
- chat_details_agent:
- choose_topic.jinja2
- system.jinja2
- table_list_agent:
- n/a
- sql_agent:
- require_join.jinja2
- select_tables.jinja2
- system.jinja2
- hvplot_agent:
- fill_plot.jinja2
- system.jinja2
- analysis_agent:
- system.jinja2
- vega_lite_agent:
- system.jinja2
I think each agent should be refactored with params like:
class ChatAgent(Agent):
purpose = param.String(
default=(
"Chats and provides info about high level data related topics, "
"e.g. what datasets are available, the columns of the data or "
"statistics about the data, and continuing the conversation. "
"Is capable of providing suggestions to get started or comment on interesting tidbits. "
"If data is available, it can also talk about the data itself."
),
doc="The directive that the Assistant will use to determine whether this agent should be selected."
)
system_prompt = param.String(
default=(
"Act as a helpful assistant for high-level data exploration, focusing on available datasets and, "
"only if data is available, explaining the purpose of each column. Offer suggestions for getting "
"started if needed, remaining factual and avoiding speculation."
)
)
prompts = param.Dict(default={"require_data": "Assess if the user's prompt requires loading data."})
response_model = param.ClassSelector(class_=BaseModel, is_instance=False)
requires = param.List(default=["current_source"], readonly=True)
It can also be paths:
class ChatAgent(Agent):
purpose = param.String(
default=str(THIS_DIR / "prompts" / "assistant" / "chat_agent.jinja2")
doc="The directive that the Assistant will use to determine whether this agent should be selected."
)
system_prompt = param.String(
default=str(THIS_DIR / "prompts" / "chat_agent" / "system.jinja2")
)
prompts = param.Dict(default={"require_data": "str(THIS_DIR / "prompts" / "chat_agent" / "require_data.jinja2")
"})
response_model = param.ClassSelector(class_=BaseModel, is_instance=False)
requires = param.List(default=["current_source"], readonly=True)
So that users can easily modify the prompts by passing a string:
ChatAgent(purpose="Choose this for chatting.", system_prompt="Talk about the data like a pirate")
Or path:
ChatAgent(purpose="path/to/assistant/chat_agent.jinja2", system_prompt="path/to/chat_agent/system.jinja2")
It would also be nice to have a Prompt
parameterized config class that contains all the built-in assistant/agents prompts, and users can use it to explore built-in prompts and update it.
class PromptsConfig(param.Parameterized):
def __init__(self, **params):
super().__init__(**params)
"""Recursively reads prompts directory and outputs a dict."""
self._prompts = ...
def dict(self) -> dict:
return self._prompts
def update(self, key: str, value: str) -> dict:
"""Updates a specific key, delimited by `.`, e.g. assistant.chat_agent or chat_agent.system"""
self._prompts[...] = value
Also, I'm starting to wonder whether each agent should only have, at most, a single invoke, i.e. split out SQLAgent into JoinAgent and QueryAgent, so that there's no need for the prompts
param.
prompts
parameterOne thing I need to think about is the response model's prompt too.
table_model = make_table_model(tables)
result = await self.llm.invoke(
messages,
system=system_prompt,
response_model=table_model,
allow_partial=False,
)
where table_model is
def make_table_model(tables):
table_model = create_model(
"Table",
chain_of_thought=(str, FieldInfo(
description="A concise, one sentence decision-tree-style analysis on choosing a table."
)),
relevant_table=(Literal[tables], FieldInfo(
description="The most relevant table based on the user query; if none are relevant, select the first."
))
)
return table_model
Which is separate from the base response model:
response = self.llm.stream(messages, system=system, response_model=Sql)
class Sql(BaseModel):
chain_of_thought: str = Field(
description="""
You are a world-class SQL expert, and your fame is on the line so don't mess up.
Then, think step by step on how you might approach this problem in an optimal way.
If it's simple, just provide one sentence.
"""
)
expr_slug: str = Field(
description="""
Give the SQL expression a concise, but descriptive, slug that includes whatever transforms were applied to it.
"""
)
query: str = Field(description="Expertly optimized, valid SQL query to be executed; do NOT add extraneous comments.")
In addition to blocks, I think we should have some pre-defined keywords available in the context to be readily injected, e.g. right now we have
system_prompt += f"\nHere are the columns of the table: {columns}"
So we can always have keywords like columns
and schema
:
import jinja2
context = {
"columns": ["a", "b", "c"],
"schema": {
"a": {"type": "enum", "objects": ["a", "A"]},
"b": {"type": "int", "min": 0, "max": 1},
"c": {"type": "float", "min": 0, "max": 1},
},
}
template = jinja2.Template(
"""
Here are the {{ columns }}.
Here are the {{ schema }}
"""
)
template.render(context).strip()
And the user can customize the template to include as much or little variables as possible.
"""
Here are the {{ columns }}.
"""
Here's my proposal for updating ChatAgent as an example (renamed system
to instructions
for clarity, since instructions + context become the system prompt of llm.invoke
)
DEFAULT_PROMPTS = {
"ChatAgent": {
"purpose": """
Chats and provides info about high level data related topics,
e.g. what datasets are available, the columns of the data or
statistics about the data, and continuing the conversation.
Is capable of providing suggestions to get started or comment on interesting tidbits.
If data is available, it can also talk about the data itself.
""",
"instructions": """
Be a helpful chatbot to talk about high-level data exploration
like what datasets are available and what each column represents.
Provide suggestions to get started if necessary. Do not write analysis
code unless explicitly asked.
""",
"context": """
{% if tables|length > 1 %}
Available tables:
{{ closest_tables }}
{% else %}
{{ table }} with schema: {{ schema }}
{% endif %}
{% if current_data %}
Here's a summary of the dataset the user just asked about:
{{ memory['current_data'] }}
```
""",
"DataRequired_instructions": """
Assess if the user's prompt requires loading data.
If the inquiry is just about available tables, no data access required.
However, if relevant tables apply to the query, load the data for a
more accurate and up-to-date response.
Here are the available tables:\n{tables_schema_str}
""",
"DataRequired_str_chain_of_thoughts": """
Thoughts on whether the user's query requires data loaded for insights;
if the user wants to explore a dataset, it's required.
If only finding a dataset, it's not required.
""",
"DataRequired_bool_data_required": """
Whether the user wants to load a specific dataset; if only searching for one, it's not required.
""",
}
}
Equivalently:
```python
DEFAULT_PROMPTS = {
"ChatAgent": {
"purpose": THIS_DIR / "prompts" / "ChatAgent_purpose.jinja2",
"instructions": THIS_DIR / "prompts" / "ChatAgent_instructions.jinja2",
"DataRequired_instructions": THIS_DIR / "prompts" / "ChatAgent_DataRequired_instructions.jinja2",
"DataRequired_str_chain_of_thoughts": THIS_DIR / "prompts" / "ChatAgent_DataRequired_str_chain_of_thoughts.jinja2",
"DataRequired_bool_data_required": THIS_DIR / "prompts" / "ChatAgent_DataRequired_bool_data_required.jinja2",
}
}
prompts = param.Dict()
, we can default to the other keys/values.For example, if user instantiates it like this without purpose
key:
ChatAgent(prompts={"instructions": "something"})
Internally, we would do
def __init__(self, **params):
prompts = params.get("prompts")
for key, value in DEFAULT_PROMPTS["ChatAgent"]:
if key not in prompts:
prompts[key] = value
params["prompts"] = prompts
prompts
param.It seems excessive, but it gives the users visibility and control into the other models within the agent's class. If we do this, then all of the existing hardcoded BaseModels will be created with create_model
dynamically.
Agent_system.jinja2
like this:{% block instructions %}
<!-- Default instructions block placeholder -->
{% endblock %}
{% block embeddings %}
<!-- Default embeddings block placeholder -->
{% endblock %}
{% block context %}
<!-- Default context block placeholder -->
{% endblock %}
Then dynamically inject
from jinja2 import Environment, FileSystemLoader
env = Environment(loader=FileSystemLoader('templates'))
base_template = env.get_template('Agent_system.jinja2')
DEFAULT_PROMPTS_TEMPLATES = {
"ChatAgent": {
"purpose": """Chats and provides info about high-level data...""",
"instructions": """Be a helpful chatbot to talk about high-level data exploration...""",
"context": """{% if tables|length > 1 %}Available tables:{{ closest_tables }}{% else %}{{ table }}{% endif %}""",
}
}
agent_name = "ChatAgent"
template_source = f"""
{{% extends 'Agent_system.jinja2' %}}
{{% block instructions %}}
{DEFAULT_PROMPTS_TEMPLATES[agent_name]['instructions']}
{{% endblock %}}
{{% block context %}}
{DEFAULT_PROMPTS_TEMPLATES[agent_name]['context']}
{{% endblock %}}
"""
dynamic_template = env.from_string(template_source)
rendered_output = dynamic_template.render(
tables=['table1', 'table2'],
closest_tables='table1, table2',
schema='schema_name'
)
print(rendered_output)
Thanks for writing that up. As I mentioned when we discussed this in person, I very strongly believe prompts should be local to the component that uses them, I do not like the idea of the global DEFAULT_PROMPTS
at all. This doesn't scale well, makes it difficult to extend and means there's multiple places to look up where a prompt comes from. Overall it just seems like a recipe for confusion.
I'm also not following some other parts of the proposal. Concretely I'm not following what DataRequired_str_chain_of_thoughts
and DataRequired_bool_data_required
are meant for? Are these the descriptions for the Model Parameters? Seems very awkward to define and I see no explanation of that naming scheme in your proposal. Divorcing the model definition from the descriptions also once again means that it's not obvious where those descriptions are coming from and it's easy for the model definition and the prompts to diverge. I'd vote for postponing a refactor of the model generation until after we're done with the prompt generation. If we want to combine these in some way we can still do that.
The templating portion of your proposal makes sense to me but to concretely say what I'd change, I'd go back to the original proposal of having a prompts
parameter on the class being the source of truth. Here's my attempt at defining a "schema" for the prompts dict:
{
<prompt-name>: {
"template": <Optional Path to base template otherwise generates dynamic template>,
"instructions": <Overrides the default instructions block>,
"context": <Overrides the default context block>"
}, ...
}
Oh and I'd probably pull out the purpose
out again. I know you didn't like using the docstring for this purpose (though I still do), so pulling it out into a purpose
parameter is fine with me.
idea of the global DEFAULT_PROMPTS
Yeah I don't mind it not being global--that was just the first idea that came to mind on how to fill out missing kwargs that the user didn't change. We can have a default_prompts class var param.
The rest sounds fine to me. I also was planning to have the prompts param, but use the other dict to fill it. I like the idea of a nested dict for prompts.
Concretely I'm not following what DataRequired_str_chain_of_thoughts and DataRequired_bool_data_required are meant for?
A way to let users customize those tool prompts as well; else it's hard coded.
Divorcing the model definition from the descriptions also once again means that it's not obvious where those descriptions are coming from
Which is why I had the naming scheme: Model name - Field Type - Field Name, but yeah we can postpone that refactor until later.
We can have a default_prompts class var param.
Fine with me, though this is a frequent pattern for us and we may want to consider whether there's something cleaner we can do about it, e.g. you might imagine we could have a special parameter type that recursively merges the class default dictionary with the user provided value. Though not today.
Model generation is definitely a more difficult problem. Since they are programmatically generated I suspect we probably want some convention for defining methods that generate the models, so by default it generates the model for a prompt by calling _<prompt-name>_model
or you can provide the concrete model class as a key to the prompts dictionary. But again, let's tackle that later.
Okay, based on our discussions this is what I came up with the following.
I think everything else, we're set on/agreed upon, except: A. The prompt name of the "base" prompt B. the paths of the jinja2 templates.
A. For the prompt name (key) of the base prompt, I'm wondering whether we should be using just base
or give it an explicit name like chat
(for ChatAgent) or write_sql
(for SQLAgent) or populate_plot
.
I'm leaning towards base
as the key for easily remembering and ease of use e.g.:
SQLAgent(prompts={"base": {"instructions": "blah"}})
vsSQLAgent.default_prompts
to figure out the key then setting SQLAgent(prompts={"write_sql": {"instructions": "blah"}})
B. For the paths, given that there's both Assistant / Coordinator and Agent, here are a few variations I'm thinking of:
For context, Assistant has these calls (with option 1 formatting):
THIS_DIR / "prompts" / "ChatAgent" / "data_required_context.jinja2"
THIS_DIR / "prompts" / "ChatAgent_data_required_context.jinja2"
THIS_DIR / "prompts" / "agents" / "ChatAgent" / "data_required_context.jinja2"
THIS_DIR / "prompts" / "agents" / "ChatAgent_data_required_context.jinja2"
THIS_DIR / "prompts" / "ChatAgent" / "data_required" / "context.jinja2"
Variation | Base Instructions Path | Base Context Path | Data Required Instructions Path | Data Required Context Path |
---|---|---|---|---|
1 | "ChatAgent" / "instructions.jinja2"| "ChatAgent" / "context.jinja2" |
"ChatAgent" / "data_required_instructions.jinja2"| "ChatAgent" / "data_required_context.jinja2" |
||
2 | "ChatAgent_instructions.jinja2"| "ChatAgent_context.jinja2" |
"ChatAgent_data_required_instructions.jinja2"| "ChatAgent_data_required_context.jinja2" |
||
3 | "agents" / "ChatAgent" / "instructions.jinja2"| "agents" / "ChatAgent" / "context.jinja2" |
"agents" / "ChatAgent" / "data_required_instructions.jinja2"| "agents" / "ChatAgent" / "data_required_context.jinja2" |
||
4 | "agents" / "ChatAgent_instructions.jinja2"| "agents" / "ChatAgent_context.jinja2" |
"agents" / "ChatAgent_data_required_instructions.jinja2"| "agents" / "ChatAgent_data_required_context.jinja2" |
||
5 | "ChatAgent" / "instructions.jinja2"| "ChatAgent" / "data_required" / "context.jinja2" |
"ChatAgent" / "data_required" / "instructions.jinja2"| "ChatAgent" / "data_required" / "context.jinja2" |
I think I prefer option 1 or 5, but not certain what's best.
Here are the implementations:
The unraveled version (where it's not just paths):
import param
from lumen.ai.assistant import Message
from lumen.ai.utils import render_template
class ChatAgent(Agent):
purpose = param.String(
default="""
Chats and provides info about high level data related topics,
e.g. what datasets are available, the columns of the data or
statistics about the data, and continuing the conversation.
Is capable of providing suggestions to get started or comment on interesting tidbits.
If data is available, it can also talk about the data itself.
"""
)
default_prompts = param.Dict(
default={
"base": {
"template": None,
"instructions": """
Be a helpful chatbot to talk about high-level data exploration
like what datasets are available and what each column represents.
Provide suggestions to get started if necessary. Do not write analysis
code unless explicitly asked.
""",
"context": """
{% if tables|length > 1 %}
Available tables:
{{ closest_tables }}
{% elseif schema %}
{{ table }} with schema: {{ schema }}
{% endif %}
{% if current_data %}
Here's a summary of the dataset the user just asked about:
{{ memory['current_data'] }}
""",
},
"data_required": {
"template": None,
"instructions": (
"Assess if the user's prompt requires loading data. "
"If the inquiry is just about available tables, no data access required. "
"However, if relevant tables apply to the query, load the data for a "
"more accurate and up-to-date response. "
),
"context": """
Here are the available tables:\n{{ tables_schema_str }}
""",
},
}
)
requires = param.List(default=["current_source"], readonly=True)
# on Agent parent class
def _render_prompt(
self, prompt_name: str, messages: list[Message], **context
) -> str:
system_prompt = render_template(
self.prompts.get(prompt_name, {}).get(
"template", self.default_prompts[prompt_name]["template"]
),
**context,
)
return system_prompt
# abstractmethod on Agent parent class
async def _render_system_prompt(self, messages: list[Message]) -> str:
source = self._memory.get("current_source")
if not source:
raise ValueError("No source found in memory.")
tables = source.get_tables()
if len(tables) > 1:
if (
len(tables) > FUZZY_TABLE_LENGTH
and "closest_tables" not in self._memory
):
closest_tables = await self._get_closest_tables(messages, tables, n=5)
else:
closest_tables = self._memory.get("closest_tables", tables)
else:
self._memory["current_table"] = table = self._memory.get(
"current_table", tables[0]
)
schema = await get_schema(self._memory["current_source"], table)
system_prompt = self._render_prompt(
"base",
messages,
tables=tables,
closest_tables=closest_tables,
table=table,
schema=schema,
current_data=self._memory.get("current_data"),
)
return system_prompt
@retry_llm_output()
async def requirements(self, messages: list[Message], errors=None):
if 'current_data' in self._memory:
return self.requires
available_sources = self._memory["available_sources"]
_, tables_schema_str = await gather_table_sources(available_sources)
system = self._render_prompt(
"data_required",
messages,
tables_schema_str=tables_schema_str,
)
with self.interface.add_step(title="Checking if data is required", steps_layout=self._steps_layout) as step:
response = self.llm.stream(
messages,
system=system,
response_model=DataRequired,
)
async for output in response:
step.stream(output.chain_of_thought, replace=True)
if output.data_required:
return self.requires + ['current_table']
return self.requires
Equivalently the paths:
```python
import param
from lumen.ai.assistant import Message
from lumen.ai.utils import render_template
class ChatAgent(Agent):
purpose = param.String(
default="""
Chats and provides info about high level data related topics,
e.g. what datasets are available, the columns of the data or
statistics about the data, and continuing the conversation.
Is capable of providing suggestions to get started or comment on interesting tidbits.
If data is available, it can also talk about the data itself.
"""
)
default_prompts = param.Dict(
default={
"base": {
"template": None,
"instructions": THIS_DIR / "prompts" / "ChatAgent" / "instructions.jinja2",
"context": THIS_DIR / "prompts" / "ChatAgent" / "context.jinja2",
},
"data_required": {
"template": None,
"instructions": THIS_DIR / "prompts" / "ChatAgent" / "data_required_instructions.jinja2",
"context": THIS_DIR / "prompts" / "ChatAgent" / "data_required_context.jinja2",
},
}
)
After this, let me know if I should go and start implementing this.
For the prompt name (key) of the base prompt, I'm wondering whether we should be using just base or give it an explicit name like chat (for ChatAgent) or write_sql (for SQLAgent) or populate_plot.
Would suggest 'main' maybe, it's not really a "base" prompt in a real sense. It's the main prompt for that agent.
B. For the paths, given that there's both Assistant / Coordinator and Agent, here are a few variations I'm thinking of:
My preference is probably:
/prompts/<agent>/<prompt>.jinja2
One thing I would suggest is that we do not break out the instructions and context by default. Instead we define them as part of one template and then let the user override that by providing their own instructions or context string or Path
, i.e. that would invert your proposal:
default_prompts = param.Dict(
default={
"base": {
"template": PROMPT_DIR / "ChatAgent" / "main.jinja2",
},
"data_required": {
"template": PROMPT_DIR / "ChatAgent" / "data_required.jinja2"
},
}
)
Having separate jinja2 files by default seems too fiddly for my liking.
Having separate jinja2 files by default seems too fiddly for my liking.
I'm actually in favor of separating the context from the instructions. Instructions should not have any dynamic variable rendered while context should. From my experience of using LangChain, it was all or nothing for changing the prompts.
For example from LangChain https://github.com/langchain-ai/langchain/blob/9611f0b55d46614a483a6789f3a034a4ead7121c/libs/langchain/langchain/agents/react/agent.py#L16:
Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}'
Say I want to change: Answer the following questions as best you can.
to Answer the following questions as best you can in a Pirate voice...
I'd have to copy & paste the entire thing because without the input/agent_scratchpad variables, it would break.
However, if it was separated into instructions and context:
instructions = """
Answer the following questions as best you can.
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
"""
context = """
{tools}
Begin!
Question: {input}
Thought:{agent_scratchpad}'''
"""
It's more manageable; I don't have to think about what the variables are; it allows focus on updating static instructions without worrying about dynamic elements.
Okay, I think this is what was intended by Philipp's comment:
from jinja2 import Environment, FileSystemLoader, DictLoader, ChoiceLoader
from markupsafe import escape
from lumen.ai.config import PROMPTS_DIR
from textwrap import dedent
# ChatAgent.default_prompts = param.Dict()
default_prompts = {
"main": {"template": PROMPTS_DIR / "ChatAgent" / "main.jinja2"},
"data_required": {"template": PROMPTS_DIR / "ChatAgent" / "data_required.jinja2"},
}
# ChatAgent.prompts = param.Dict()
prompts = {}
# Determine the base template path
template_path = default_prompts["main"]["template"].relative_to(PROMPTS_DIR)
fs_loader = FileSystemLoader(PROMPTS_DIR)
if prompts:
# Dynamically create block definitions based on dictionary keys with proper escaping
block_definitions = "\n".join(
f"{{% block {escape(key)} %}}{escape(value)}{{% endblock %}}"
for key, value in prompts.items()
)
# Create the dynamic template content by extending the base template and adding blocks
dynamic_template_content = dedent(
f"""
{{% extends "{template_path}" %}}
{block_definitions}
"""
)
dynamic_loader = DictLoader({"dynamic_template": dynamic_template_content})
choice_loader = ChoiceLoader([dynamic_loader, fs_loader])
env = Environment(loader=choice_loader)
# Load the dynamic template
template = env.get_template("dynamic_template")
else:
env = Environment(loader=fs_loader)
template = env.get_template(str(template_path))
rendered_output = template.render(
tables=["table1", "table2"],
closest_tables="table1, table2",
schema="schema_name",
memory={"current_data": "abc"},
)
print(rendered_output)
If prompts = {}
:
If prompts
is set with custom instructions
Now I'm wondering whether we need the nesting (main -> template -> path):
e.g. should it be renamed to default_prompt_templates
default_prompt_templates = param.Dict(
default={
"main": PROMPT_DIR / "ChatAgent" / "main.jinja2",
"data_required": PROMPT_DIR / "ChatAgent" / "data_required.jinja2",
}
)
and prompts
be renamed to prompt_overrides
?
At the current moment, there are multiple entry points for prompting the LLM
BaseModel
sBaseModel
sI probably am missing one, but I propose that for maximum customizability, we should migrate all of these to jinja2 templates, and dynamically read the prompts from the jinja2 templates and create models dynamically too.
That way, users can update the prompts to their liking by either passing in
system_prompt_paths
with the corresponding names or updating aSYSTEM_PROMPTS[key]
dict.We should also make it easy for users to discover these prompts.