Using Guidance with AutoGen
This notebook shows how Guidance can be used to enable structured responses from AutoGen agents. In particular, this notebook focuses on creating agents that always output a valid code block or valid json object.
import re
from guidance import assistant, gen, models, system, user
from pydantic import BaseModel
from autogen import Agent, AssistantAgent, UserProxyAgent, config_list_from_json
llm_config = config_list_from_json("OAI_CONFIG_LIST")[0] # use the first config
gpt = models.OpenAI("gpt-4", api_key=llm_config.get("api_key"))
The example below uses guidance to create a guidance_coder agent that
only responds with valid code blocks.
def is_valid_code_block(code):
pattern = r"```[\w\s]*\n([\s\S]*?)\n```"
match = re.search(pattern, code)
if match:
return True
else:
return False
def generate_structured_response(recipient, messages, sender, config):
gpt = models.OpenAI("gpt-4", api_key=llm_config.get("api_key"), echo=False)
# populate the recipient with the messages from the history
with system():
lm = gpt + recipient.system_message
for message in messages:
if message.get("role") == "user":
with user():
lm += message.get("content")
else:
with assistant():
lm += message.get("content")
# generate a new response and store it
with assistant():
lm += gen(name="initial_response")
# ask the agent to reflect on the nature of the response and store it
with user():
lm += "Does the very last response from you contain code? Respond with yes or no."
with assistant():
lm += gen(name="contains_code")
# if the response contains code, ask the agent to generate a proper code block
if "yes" in lm["contains_code"].lower():
with user():
lm += "Respond with a single blocks containing the valid code. Valid code blocks start with ```"
with assistant():
lm += "```" + gen(name="code")
response = "```" + lm["code"]
is_valid = is_valid_code_block(response)
if not is_valid:
raise ValueError(f"Failed to generate a valid code block\n {response}")
# otherwise, just use the initial response
else:
response = lm["initial_response"]
return True, response
guidance_agent = AssistantAgent("guidance_coder", llm_config=llm_config)
guidance_agent.register_reply(Agent, generate_structured_response, 1)
user_proxy = UserProxyAgent(
"user",
human_input_mode="TERMINATE",
code_execution_config={
"work_dir": "coding",
"use_docker": False,
}, # Please set use_docker=True if docker is available to run the generated code. Using docker is safer than running the generated code directly.
is_termination_msg=lambda msg: "TERMINATE" in msg.get("content"),
)
user_proxy.initiate_chat(guidance_agent, message="Plot and save a chart of nvidia and tsla stock price change YTD.")
user (to guidance_coder):
Plot and save a chart of nvidia and tsla stock price change YTD.
--------------------------------------------------------------------------------
guidance_coder (to user):
```python
# filename: stock_price_change.py
import pandas as pd
import yfinance as yf
import matplotlib.pyplot as plt
from datetime import datetime
# Get today's date
today = datetime.today().strftime('%Y-%m-%d')
# Download stock data
nvda = yf.download('NVDA', start='2022-01-01', end=today)
tsla = yf.download('TSLA', start='2022-01-01', end=today)
# Calculate percentage change in closing price
nvda['Pct Change'] = nvda['Close'].pct_change()
tsla['Pct Change'] = tsla['Close'].pct_change()
# Plot percentage change
plt.figure(figsize=(14,7))
plt.plot(nvda['Pct Change'], label='NVDA')
plt.plot(tsla['Pct Change'], label='TSLA')
plt.title('Nvidia and Tesla Stock Price Change YTD')
plt.xlabel('Date')
plt.ylabel('Percentage Change')
plt.legend()
plt.grid(True)
# Save the plot as a PNG file
plt.savefig('stock_price_change.png')
```
--------------------------------------------------------------------------------
>>>>>>>> USING AUTO REPLY...
>>>>>>>> EXECUTING CODE BLOCK 0 (inferred language is python)...
user (to guidance_coder):
exitcode: 0 (execution succeeded)
Code output:
[*********************100%%**********************] 1 of 1 completed
[*********************100%%**********************] 1 of 1 completed
--------------------------------------------------------------------------------
guidance_coder (to user):
Great! The code executed successfully and the chart of Nvidia and Tesla stock price change Year-To-Date (YTD) has been saved as 'stock_price_change.png' in the current directory. You can open this file to view the chart.
TERMINATE
--------------------------------------------------------------------------------
execute_code was called without specifying a value for use_docker. Since the python docker package is not available, code will be run natively. Note: this fallback behavior is subject to change
The example below uses Guidance to enable a guidance_labeler agent
that only responds with a valid JSON that labels a given comment/joke.
class Response(BaseModel):
label: str
explanation: str
response_prompt_instructions = """The label must be a JSON of the format:
{
"label": str,
"explanation": str
}"""
def generate_structured_response(recipient, messages, sender, config):
gpt = models.OpenAI("gpt-4", api_key=llm_config.get("api_key"), echo=False)
# populate the recipient with the messages from the history
with system():
lm = gpt + recipient.system_message
for message in messages:
if message.get("role") == "user":
with user():
lm += message.get("content")
else:
with assistant():
lm += message.get("content")
# generate a new response and store it
with assistant():
lm += gen(name="initial_response")
# ask the agent to reflect on the nature of the response and store it
with user():
lm += "Does the very last response from you contain JSON object? Respond with yes or no."
with assistant():
lm += gen(name="contains_json")
# if the response contains code, ask the agent to generate a proper code block
if "yes" in lm["contains_json"].lower():
with user():
lm += (
"What was that JSON object? Only respond with that valid JSON string. A valid JSON string starts with {"
)
with assistant():
lm += "{" + gen(name="json")
response = "{" + lm["json"]
# verify that the response is valid json
try:
response_obj = Response.model_validate_json(response)
response = response_obj.model_dump_json()
except Exception as e:
response = str(e)
# otherwise, just use the initial response
else:
response = lm["initial_response"]
return True, response
guidance_agent = AssistantAgent("guidance_labeler", llm_config=llm_config, system_message="You are a helpful assistant")
guidance_agent.register_reply(Agent, generate_structured_response, 1)
user_proxy = UserProxyAgent("user", human_input_mode="ALWAYS", code_execution_config=False)
user_proxy.initiate_chat(
guidance_agent,
message=f"""
Label the TEXT via the following instructions:
{response_prompt_instructions}
TEXT: what did the fish say when it bumped into a wall? Dam!
""",
)
user (to guidance_labeler):
Label the TEXT via the following instructions:
The label must be a JSON of the format:
{
"label": str,
"explanation": str
}
TEXT: what did the fish say when it bumped into a wall? Dam!
--------------------------------------------------------------------------------
guidance_labeler (to user):
{"label":"Joke","explanation":"The text is a joke, using a play on words where the fish says 'Dam!' after bumping into a wall, which is a pun on the word 'damn' and a 'dam' which is a barrier that stops or restricts the flow of water, often creating a reservoir, and is something a fish might encounter."}
--------------------------------------------------------------------------------