Using Guidance with AutoGen

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"))

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)...

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

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

--------------------------------------------------------------------------------

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."}

--------------------------------------------------------------------------------