Not a pizza but a Crème Br?lée with LLM caramelized crust
Jenya Stoeva
Product @ heart / No man is an island, entire of itself; Every man is a piece of the continent, a part of the main. /
This isn't about using Anthropic Claude Sonnet 3.5 to order a pizza through a GUI, nor is it about making a Crème Br?lée—although... keep reading.?It is about an idea I felt equally excited cooking up.
I've long imagined being able to seamlessly connect to various tools through a single interface to query data on the fly. Envision simply writing a free-form query and receiving the data without having to determine the appropriate API endpoints or handle authentication complexities.
I am excited to share my exploratory journey and beware I am not an engineer!?It's fascinating to witness the paradox of LLMs: while they have huge limitations, they empower us to accomplish things that once seemed beyond reach.
The project utilizes a sequence of LangChain templates that contain the model's instructions (prompts) and uses Jupyter Notebook as the execution environment. Let’s dive in!
Given the free-form query, "Get all Jira issues in project EMVP for which a comment has been added in the past week." prompt an LLM ( OpenAI GPT-4o) to generate valid Python code that connects to Jira and retrieves the specified data. The output from each step can be seen at the end.
Step 1 :: Transform Free-Form Input into Code
My strategy was to transition to Python code as swiftly as possible and refine it through iterative improvements until achieving the desired outcome. Why?
Python is now?the top programming language on GitHub?https://lnkd.in/gmTiJJJx.?Models have been?extensively?trained on Python code.?
The top 3 most successful tasks that a Large Language Model (LLM) can complete are:
Step 2 :: Validate Python Code
There are various methods to approach validation, but I chose the simplest one: passing the generated Python code back to the same model for review and correction.
Step 3 :: Map Environment Variables for Successful Authentication
Step 4 :: Execute Python Code
In this step, the Python code from Step 3 is executed for the first time. The code attempts to establish a connection to Jira and retrieve the specified data. The outcome of this execution is captured as the Response.
Step 5 :: Test (the best step)
This step is where the model thoroughly reviews and corrects any issues in the Python code. It takes the executed Python code, the response from the initial run, and the original user input from Step 1. The model then analyzes these components to determine if the code successfully achieves the intended goal and makes any necessary adjustments.
领英推荐
Step 6 :: Final Execution
Will this approach work consistently and reliably in production? Probably not—but that's for the engineers to determine, not me. What I do believe, though, is that as the models advance and the Python ecosystem continues to evolve, it just might work in the future.
Until then, keep dreaming, learning, and experimenting—and now, treat yourself to a well-deserved Crème Br?lée. This is how it was prepared by GPT-4o with a little help from me, using the method I described above. ??
If you enjoyed this journey, don’t forget to give it a thumbs up!
Coming up next is the UI and I will appreciate any help with Python!
If you have missed how to order a pizza with the help of Claude Sonnet 3.5, check the comments.
Step 1
To address the request for retrieving Jira issues with recent comments, we will use the `jira` Python library, which is a dedicated library for interacting with the Jira API. This library simplifies authentication and API requests. Below is the Python code to achieve this, followed by a JSON object listing the variable elements.
### Produced Code
```python
from jira import JIRA
import logging
from io import StringIO
from datetime import datetime, timedelta
# Set up a string stream to capture log messages
log_stream = StringIO()
logging.basicConfig(level=logging.INFO, stream=log_stream, format='%(levelname)s:%(message)s', force=True)
# Define variables for the connection
jira_server = "https://your-jira-instance.atlassian.net" # Replace with your Jira server URL
jira_username = "[email protected]" # Replace with your Jira username (usually an email)
jira_api_token = "your-api-token" # Replace with your Jira API token
project_key = "EMVP" # Replace with your project key
# Initialize the JIRA connection
def connect_to_jira():
try:
options = {'server': jira_server}
jira = JIRA(options, basic_auth=(jira_username, jira_api_token))
return jira
except Exception as e:
logging.error("Failed to connect to Jira: %s", str(e))
return None
# Fetch issues with comments added in the past week
def get_recent_commented_issues(jira):
try:
# Calculate the date one week ago
one_week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
# JQL query to find issues with comments added in the past week
jql_query = f'project = {project_key} AND updated >= "{one_week_ago}" AND comment is not EMPTY'
# Search for issues
issues = jira.search_issues(jql_query)
# Log the issues found
for issue in issues:
logging.info("Issue found: %s - %s", issue.key, issue.fields.summary)
except Exception as e:
logging.error("Failed to fetch issues: %s", str(e))
# Main workflow
jira_client = connect_to_jira()
if jira_client:
get_recent_commented_issues(jira_client)
else:
logging.error("Jira connection could not be established.")
# Flush the log stream to ensure all messages are captured
log_stream.flush()
# Return the log messages
log_messages = log_stream.getvalue()
print(log_messages)
```
### Output as a JSON Object
```json
{
"jira_server": "string",
"jira_username": "string",
"jira_api_token": "string",
"project_key": "string"
}
```
This code connects to a Jira instance using basic authentication with an API token, searches for issues in a specified project that have had comments added in the past week, and logs the results. The variable elements are extracted and listed in the JSON object for easy configuration.
Step 2
The provided code for connecting to Jira and retrieving issues with comments added in the past week looks mostly correct, but let's review it for any potential errors or improvements:
1. **Imports**: The necessary libraries are imported, including `JIRA` from the `jira` package, `logging`, `StringIO` from `io`, and date-related classes from `datetime`.
2. **Authentication**: The code uses basic authentication with a username and API token, which is appropriate for Jira Cloud instances. Ensure that the `jira` package is installed (`pip install jira`).
3. **Error Handling**: The code includes error handling using try-except blocks, which is good practice.
4. **Logging**: The logging setup captures log messages into a `StringIO` stream, which is correctly implemented.
5. **JQL Query**: The JQL query seems correct for fetching issues updated in the past week with comments.
6. **Variable Definitions**: Ensure that the variables `jira_server`, `jira_username`, `jira_api_token`, and `project_key` are correctly populated with your Jira instance details.
7. **Functionality**: The code should work as intended if all variables are correctly set.
Here are a few suggestions for improvement:
- **Security**: Ensure that sensitive information like API tokens is stored securely, possibly using environment variables or a secure vault.
- **Pagination**: If there are many issues, consider handling pagination in the `search_issues` method to ensure all issues are retrieved.
- **Logging Level**: You might want to adjust the logging level based on the environment (e.g., DEBUG for development, INFO or WARNING for production).
- **Date Handling**: Ensure the date format in the JQL query matches your Jira instance's expected format.
Here's the final code with a minor improvement for pagination:
```python
from jira import JIRA
import logging
from io import StringIO
from datetime import datetime, timedelta
# Set up a string stream to capture log messages
log_stream = StringIO()
logging.basicConfig(level=logging.INFO, stream=log_stream, format='%(levelname)s:%(message)s', force=True)
# Define variables for the connection
jira_server = "https://your-jira-instance.atlassian.net" # Replace with your Jira server URL
jira_username = "[email protected]" # Replace with your Jira username (usually an email)
jira_api_token = "your-api-token" # Replace with your Jira API token
project_key = "EMVP" # Replace with your project key
# Initialize the JIRA connection
def connect_to_jira():
try:
options = {'server': jira_server}
jira = JIRA(options, basic_auth=(jira_username, jira_api_token))
return jira
except Exception as e:
logging.error("Failed to connect to Jira: %s", str(e))
return None
# Fetch issues with comments added in the past week
def get_recent_commented_issues(jira):
try:
# Calculate the date one week ago
one_week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
# JQL query to find issues with comments added in the past week
jql_query = f'project = {project_key} AND updated >= "{one_week_ago}" AND comment is not EMPTY'
# Search for issues with pagination
start_at = 0
max_results = 50
while True:
issues = jira.search_issues(jql_query, startAt=start_at, maxResults=max_results)
if not issues:
break
for issue in issues:
logging.info("Issue found: %s - %s", issue.key, issue.fields.summary)
start_at += max_results
except Exception as e:
logging.error("Failed to fetch issues: %s", str(e))
# Main workflow
jira_client = connect_to_jira()
if jira_client:
get_recent_commented_issues(jira_client)
else:
logging.error("Jira connection could not be established.")
# Flush the log stream to ensure all messages are captured
log_stream.flush()
# Return the log messages
log_messages = log_stream.getvalue()
print(log_messages)
```
**JSON Object for Variables**:
```json
{
"jira_server": "string",
"jira_username": "string",
"jira_api_token": "string",
"project_key": "string"
}
```
Ensure that you replace the placeholders with your actual Jira instance details and credentials.
?Step 3
To update the code with environment variables, we need to map the existing variables to the appropriate environment variables based on their context. Here's how the code should look after the substitution:
```python
import os
from jira import JIRA
import logging
from io import StringIO
from datetime import datetime, timedelta
# Set up a string stream to capture log messages
log_stream = StringIO()
logging.basicConfig(level=logging.INFO, stream=log_stream, format='%(levelname)s:%(message)s', force=True)
# Define variables for the connection using environment variables
jira_server = os.getenv('JIRA_URL') # Replace with your Jira server URL
jira_username = os.getenv('JIRA_EMAIL') # Replace with your Jira username (usually an email)
jira_api_token = os.getenv('JIRA_API_TOKEN') # Replace with your Jira API token
project_key = "EMVP" # Replace with your project key
# Initialize the JIRA connection
def connect_to_jira():
try:
options = {'server': jira_server}
jira = JIRA(options, basic_auth=(jira_username, jira_api_token))
return jira
except Exception as e:
logging.error("Failed to connect to Jira: %s", str(e))
return None
# Fetch issues with comments added in the past week
def get_recent_commented_issues(jira):
try:
# Calculate the date one week ago
one_week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
# JQL query to find issues with comments added in the past week
jql_query = f'project = {project_key} AND updated >= "{one_week_ago}" AND comment is not EMPTY'
# Search for issues with pagination
start_at = 0
max_results = 50
while True:
issues = jira.search_issues(jql_query, startAt=start_at, maxResults=max_results)
if not issues:
break
for issue in issues:
logging.info("Issue found: %s - %s", issue.key, issue.fields.summary)
start_at += max_results
except Exception as e:
logging.error("Failed to fetch issues: %s", str(e))
# Main workflow
jira_client = connect_to_jira()
if jira_client:
get_recent_commented_issues(jira_client)
else:
logging.error("Jira connection could not be established.")
# Flush the log stream to ensure all messages are captured
log_stream.flush()
# Return the log messages
log_messages = log_stream.getvalue()
print(log_messages)
```
In this updated code, the `jira_server`, `jira_username`, and `jira_api_token` variables are now set using environment variables `JIRA_URL`, `JIRA_EMAIL`, and `JIRA_API_TOKEN`, respectively. The `project_key` remains hardcoded as "EMVP" since it wasn't mapped to an environment variable in your request.
Step 4
import sys
import json
import requests
from io import StringIO
# Execute the provided Python code safely
try:
# Capture the output of the executed code
exec_locals = {}
exec(execute_python_code_updated_env_vars, globals(), exec_locals)
# Retrieve log messages from the local variables
if 'log_messages' in exec_locals:
response_variable = exec_locals['log_messages']
else:
response_variable = "No log messages captured."
except Exception as e:
response_variable = f"Execution failed: {str(e)}"
# Print or return the log messages for further processing
print(response_variable)
ERROR:Failed to fetch issues: JiraError HTTP 400 url: https://finstep.atlassian.net/rest/api/2/search?jql=project+%3D+EMVP+AND+updated+%3E%3D+%222024-10-31%22+AND+comment+is+not+EMPTY&startAt=0&validateQuery=True&fields=%2Aall&maxResults=50
text: The field 'comment' does not support searching for EMPTY values.
response headers = {'Date': 'Thu, 07 Nov 2024 12:34:33 GMT', 'Content-Type': 'application/json;charset=UTF-8', 'Server': 'AtlassianEdge', 'Timing-Allow-Origin': '*', 'X-Arequestid': 'e607c2779c50a3ab14dbc792a50c0f59', 'X-Aaccountid': '6273ca48d7fd480068d98038', 'Cache-Control': 'no-cache, no-store, no-transform', 'X-Content-Type-Options': 'nosniff', 'X-Xss-Protection': '1; mode=block', 'Atl-Traceid': 'c3b7a8c722434828b10c26e7d404664f', 'Atl-Request-Id': 'c3b7a8c7-2243-4828-b10c-26e7d404664f', 'Strict-Transport-Security': 'max-age=63072000; includeSubDomains; preload', 'Report-To': '{"endpoints": [{"url": "https://dz8aopenkvv6s.cloudfront.net"}], "group": "endpoint-1", "include_subdomains": true, "max_age": 600}', 'Nel': '{"failure_fraction": 0.001, "include_subdomains": true, "max_age": 600, "report_to": "endpoint-1"}', 'Server-Timing': 'atl-edge;dur=218,atl-edge-internal;dur=12,atl-edge-upstream;dur=206,atl-edge-pop;desc="aws-eu-central-1"', 'Transfer-Encoding': 'chunked'}
response text = {"errorMessages":["The field 'comment' does not support searching for EMPTY values."],"warningMessages":[]}
Step 5
The error message indicates that the JQL query is incorrect because the `comment` field does not support searching for `EMPTY` values. To fix this, you can modify the JQL query to check for issues updated in the past week and then filter them in Python to see if they have comments added in the past week.
Here's the corrected code:
```python
import requests
from pathlib import Path
from dotenv import load_dotenv, find_dotenv, dotenv_values
import os
from jira import JIRA
import logging
from io import StringIO
from datetime import datetime, timedelta
# Define the path to the environment file
env_file_path = Path("oneConnection.env")
# Load environment variables from the specific file
load_dotenv(env_file_path, override=True) # This loads variables into os.environ
# Set up a string stream to capture log messages
log_stream = StringIO()
logging.basicConfig(level=logging.INFO, stream=log_stream, format='%(levelname)s:%(message)s', force=True)
# Define variables for the connection using environment variables
jira_server = os.getenv('JIRA_URL') # Replace with your Jira server URL
jira_username = os.getenv('JIRA_EMAIL') # Replace with your Jira username (usually an email)
jira_api_token = os.getenv('JIRA_API_TOKEN') # Replace with your Jira API token
project_key = "EMVP" # Replace with your project key
# Initialize the JIRA connection
def connect_to_jira():
try:
options = {'server': jira_server}
jira = JIRA(options, basic_auth=(jira_username, jira_api_token))
return jira
except Exception as e:
logging.error("Failed to connect to Jira: %s", str(e))
return None
# Fetch issues updated in the past week and check for recent comments
def get_recent_commented_issues(jira):
try:
# Calculate the date one week ago
one_week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
# JQL query to find issues updated in the past week
jql_query = f'project = {project_key} AND updated >= "{one_week_ago}"'
# Search for issues with pagination
start_at = 0
max_results = 50
while True:
issues = jira.search_issues(jql_query, startAt=start_at, maxResults=max_results)
if not issues:
break
for issue in issues:
# Check if the issue has comments added in the past week
comments = jira.comments(issue)
for comment in comments:
comment_created = datetime.strptime(comment.created.split('.')[0], '%Y-%m-%dT%H:%M:%S')
if comment_created >= datetime.now() - timedelta(days=7):
logging.info("Issue found with recent comment: %s - %s", issue.key, issue.fields.summary)
break
start_at += max_results
except Exception as e:
logging.error("Failed to fetch issues: %s", str(e))
# Main workflow
jira_client = connect_to_jira()
if jira_client:
get_recent_commented_issues(jira_client)
else:
logging.error("Jira connection could not be established.")
# Flush the log stream to ensure all messages are captured
log_stream.flush()
# Return the log messages
log_messages = log_stream.getvalue()
print(log_messages)
```
### Key Changes:
1. **JQL Query Update**: The JQL query now only checks for issues updated in the past week without filtering by comments.
2. **Comment Filtering in Python**: After fetching the issues, the code checks each issue's comments to see if any were added in the past week.
3. **Datetime Parsing**: The `comment.created` field is parsed to a `datetime` object to compare with the current date minus seven days.
This should resolve the error and correctly identify issues with comments added in the past week.
Step 6
import requests
from pathlib import Path
from dotenv import load_dotenv, find_dotenv, dotenv_values
import os
from jira import JIRA
import logging
from io import StringIO
from datetime import datetime, timedelta
# Define the path to the environment file
env_file_path = Path("oneConnection.env")
# Load environment variables from the specific file
load_dotenv(env_file_path, override=True) # This loads variables into os.environ
# Set up a string stream to capture log messages
log_stream = StringIO()
logging.basicConfig(level=logging.INFO, stream=log_stream, format='%(levelname)s:%(message)s', force=True)
# Define variables for the connection using environment variables
jira_server = os.getenv('JIRA_URL') # Replace with your Jira server URL
jira_username = os.getenv('JIRA_EMAIL') # Replace with your Jira username (usually an email)
jira_api_token = os.getenv('JIRA_API_TOKEN') # Replace with your Jira API token
project_key = "EMVP" # Replace with your project key
# Initialize the JIRA connection
def connect_to_jira():
try:
options = {'server': jira_server}
jira = JIRA(options, basic_auth=(jira_username, jira_api_token))
return jira
except Exception as e:
logging.error("Failed to connect to Jira: %s", str(e))
return None
# Fetch issues updated in the past week and check for recent comments
def get_recent_commented_issues(jira):
try:
# Calculate the date one week ago
one_week_ago = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d')
# JQL query to find issues updated in the past week
jql_query = f'project = {project_key} AND updated >= "{one_week_ago}"'
# Search for issues with pagination
start_at = 0
max_results = 50
while True:
issues = jira.search_issues(jql_query, startAt=start_at, maxResults=max_results)
if not issues:
break
for issue in issues:
# Check if the issue has comments added in the past week
comments = jira.comments(issue)
for comment in comments:
comment_created = datetime.strptime(comment.created.split('.')[0], '%Y-%m-%dT%H:%M:%S')
if comment_created >= datetime.now() - timedelta(days=7):
logging.info("Issue found with recent comment: %s - %s", issue.key, issue.fields.summary)
break
start_at += max_results
except Exception as e:
logging.error("Failed to fetch issues: %s", str(e))
# Main workflow
jira_client = connect_to_jira()
if jira_client:
get_recent_commented_issues(jira_client)
else:
logging.error("Jira connection could not be established.")
# Flush the log stream to ensure all messages are captured
log_stream.flush()
# Return the log messages
log_messages = log_stream.getvalue()
print(log_messages)
INFO:Issue found with recent comment: EMVP-735 - Tracking Onboarding success rate
INFO:Issue found with recent comment: EMVP-734 - Many SVGs are improperly scaled on phone or ipad, but not in the simulator
INFO:Issue found with recent comment: EMVP-733 - lecture step points are not credited
INFO:Issue found with recent comment: EMVP-730 - Notifications on life replenish
This is the perfect Crème Br?lée article to find for anyone asking about Crème Br?lée
Advisory
3 个月great write up love it
Product @ heart / No man is an island, entire of itself; Every man is a piece of the continent, a part of the main. /
3 个月And if you wish to order a pizza instead https://www.dhirubhai.net/posts/mikegchambers_i-gave-claude-sonnet-access-to-control-my-activity-7259540504072134656-7EsK?utm_source=share&utm_medium=member_desktop