Created
February 15, 2025 22:46
-
-
Save rkorszun/eee513863bb6217b47a3dc99a6cae13b to your computer and use it in GitHub Desktop.
Example of using deepseek (ollama - local) with some external API using json_schema
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from pydantic import BaseModel, Field | |
from datetime import date | |
import ollama | |
import requests | |
import json | |
import pandas as pd | |
class Query(BaseModel): | |
"""Data that we want collect from user to get weather data""" | |
place: str = Field(description="The city/address/POI where the temperature is being asked for") | |
countryCode: str = Field(description="The country ISO code where is being asked for") | |
start_date: date = Field(description="The start date of asked period if not detected the same as end date") | |
end_date: date = Field(description="The start date of asked period") | |
def get_weather_data(user_input: str) -> dict: | |
response = ollama.chat( | |
model='deepseek-r1:7b', | |
messages=[ | |
{'role': 'user', 'content': 'get basic structure of this query:' + user_input}, | |
], | |
format=Query.model_json_schema() | |
) | |
return json.loads(response['message']['content']) | |
def ask_for_missing_properties(weather_data: dict) -> str: | |
response = ollama.chat( | |
model='deepseek-r1:7b', | |
messages=[ | |
{'role': 'user', 'content': 'Ask for missing information :' + str(weather_data)}, | |
], | |
) | |
return response['message']['content'] | |
def create_summary(user_input: str, info: dict) -> str: | |
response = ollama.chat( | |
model='deepseek-r1:7b', | |
messages=[ | |
{'role': 'user', 'content': 'answer for question :' + user_input + ' using this data in one short sentence' + str(info)}, | |
], | |
) | |
resp = response['message']['content'] | |
# remove all befoe '</think>' tag | |
return resp.split('</think>')[1].strip() | |
def get_location_info(weather_data: dict) -> dict: | |
query = weather_data['place'] + "," + weather_data['countryCode'] | |
url = "https://nominatim.openstreetmap.org/search" | |
params = { | |
'q': query, | |
'format': 'json', | |
'limit': 1 | |
} | |
response = requests.get(url, params=params, headers={'User-Agent': 'Mozilla/5.0'}) | |
return response.json()[0] | |
def get_weather_info(location: dict, weather_data: dict) -> dict: | |
url = "https://archive-api.open-meteo.com/v1/archive" | |
params = { | |
'latitude': location['lat'], | |
'longitude': location['lon'], | |
'start_date': weather_data['start_date'], | |
'end_date': weather_data['end_date'], | |
'hourly': 'temperature_2m', | |
'timezone': 'auto' | |
} | |
response = requests.get(url, params=params, headers={'User-Agent': 'Mozilla/5.0'}) | |
data = response.json()['hourly'] | |
df = pd.DataFrame(data) | |
df['date'] = pd.to_datetime(df['time']).dt.date | |
df = df.groupby('date').agg({'temperature_2m': ['min', 'max']}) | |
return df.to_dict() | |
if __name__ == "__main__": | |
user_input = input("Ask for some weather data: ") | |
weather_data = get_weather_data(user_input) | |
print(weather_data) | |
location = get_location_info(weather_data) | |
info = get_weather_info(location, weather_data) | |
print(create_summary(user_input, info)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This is using ollama with
ollama pull deepseek-r1:7b