mirror of
https://github.com/storytold/artcraft.git
synced 2026-11-05 01:38:44 +00:00
Add new status columns to list queries
This commit is contained in:
1 parent
b511420c03
commit
95f2ec3e40
20 files changed
+561
-7
No files matched your search
+8
@@ -20,8 +20,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+8
@@ -20,8 +20,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+8
@@ -20,8 +20,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+8
@@ -25,8 +25,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+8
@@ -25,8 +25,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+8
@@ -20,8 +20,16 @@ SELECT
|
||||
web_content_type as `web_content_type: enums::common::sqlite::web_content_type::WebContentType`,
|
||||
original_news_title,
|
||||
overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
|
||||
llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_rendition_attempts,
|
||||
|
||||
llm_title_summary_status as `llm_title_summary_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_title_summary_attempts,
|
||||
|
||||
llm_categorization_status as `llm_categorization_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
llm_categorization_attempts,
|
||||
|
||||
audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_status as `image_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
|
||||
image_generation_attempts
|
||||
|
||||
+9
-1
@@ -12,10 +12,18 @@ pub struct NewsStoryProductionItem {
|
||||
// Overall production status, which depends on all other statuses
|
||||
pub overall_production_status: AwaitableJobStatus,
|
||||
|
||||
// LLM rendition task
|
||||
// LLM main text rendition task
|
||||
pub llm_rendition_status: AwaitableJobStatus,
|
||||
pub llm_rendition_attempts: i64,
|
||||
|
||||
// LLM title summary task
|
||||
pub llm_title_summary_status: AwaitableJobStatus,
|
||||
pub llm_title_summary_attempts: i64,
|
||||
|
||||
// LLM categorization task
|
||||
pub llm_categorization_status: AwaitableJobStatus,
|
||||
pub llm_categorization_attempts: i64,
|
||||
|
||||
// Audio generation tasks (farmed out to 1:n work items)
|
||||
pub audio_generation_status: AwaitableJobStatus,
|
||||
|
||||
|
||||
+3
-3
@@ -22,7 +22,7 @@ CREATE TABLE news_story_productions (
|
||||
maybe_summary_news_title TEXT,
|
||||
|
||||
-- Comma-separated categories for the story. Generated by LLM.
|
||||
maybe_summary_categories TEXT,
|
||||
maybe_categorization TEXT,
|
||||
|
||||
-- If we skip the production at any stage (after scraping, after rendition, etc.), for a non-failure
|
||||
-- reason, this is why.
|
||||
@@ -49,10 +49,10 @@ CREATE TABLE news_story_productions (
|
||||
|
||||
-- Classifying the categories with LLM/GPT
|
||||
-- "not_ready", "ready_waiting", "processing", "retryably_failed", "permanently_failed", "skipped", "done"
|
||||
llm_category_summary_status TEXT DEFAULT "not_ready" NOT NULL,
|
||||
llm_categorization_status TEXT DEFAULT "not_ready" NOT NULL,
|
||||
|
||||
-- Number of LLM/GPT category summary attempts that have been made
|
||||
llm_category_summary_attempts INT DEFAULT 0 NOT NULL,
|
||||
llm_categorization_attempts INT DEFAULT 0 NOT NULL,
|
||||
|
||||
-- TTS for the audio with FakeYou
|
||||
-- "not_ready", "ready_waiting", "processing", "retryably_failed", "permanently_failed", "skipped", "done"
|
||||
|
||||
+2
-2
@@ -19,10 +19,10 @@ CREATE TABLE news_stories (
|
||||
original_news_title TEXT NOT NULL,
|
||||
|
||||
-- LLM renditioned news title, short. Under 60 characters.
|
||||
--summary_news_title TEXT NOT NULL,
|
||||
summary_news_title TEXT NOT NULL,
|
||||
|
||||
-- Comma-separated categories for the story. Generated by LLM.
|
||||
--summary_categories TEXT NOT NULL,
|
||||
llm_categorization TEXT NOT NULL,
|
||||
|
||||
-- Total number of discrete audio files for this story
|
||||
audio_file_count INTEGER NOT NULL,
|
||||
|
||||
@@ -1,2 +1,4 @@
|
||||
pub mod news_story_llm_rendition;
|
||||
pub mod news_story_image_generation;
|
||||
pub mod news_story_llm_category_summary;
|
||||
pub mod news_story_llm_rendition;
|
||||
pub mod news_story_llm_title_summary;
|
||||
+1
@@ -0,0 +1 @@
|
||||
pub mod news_article_prompt;
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
|
||||
// TODO: We'll add more types of prompt templates in the future.
|
||||
|
||||
pub struct NewsArticlePrompt {
|
||||
// TODO: Add other details (title, etc.)
|
||||
paragraphs: Vec<String>,
|
||||
}
|
||||
|
||||
impl NewsArticlePrompt {
|
||||
|
||||
pub fn new(paragraphs: &Vec<String>) -> Self {
|
||||
Self {
|
||||
paragraphs: paragraphs.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: This will need a lot of work and experimentation.
|
||||
pub fn make_prompt(&self) -> String {
|
||||
let start = "The following is a news article:\n\n";
|
||||
|
||||
let end = "\n\nPlease reword the above article as if you were a newscaster presenting the news. Spell out any abbreviations and numbers";
|
||||
|
||||
format!("{}{}{}", start, &self.paragraphs.join("\n\n"), end)
|
||||
}
|
||||
}
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
use crate::shared_state::job_state::JobState;
|
||||
use crate::workers::news_stories::phase_2::news_story_llm_rendition::process_single_item::process_single_item;
|
||||
use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus;
|
||||
use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
|
||||
use log::{debug, error, info};
|
||||
use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_llm_rendition::list_news_story_productions_awaiting_llm_rendition;
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
/// Project scraped, summarize the categories.
|
||||
pub async fn news_story_llm_category_summary_main_loop(job_state: Arc<JobState>) {
|
||||
loop {
|
||||
debug!("news_story_llm_category_summary_main_loop main loop");
|
||||
|
||||
single_job_loop_iteration(&job_state).await;
|
||||
|
||||
debug!("llm_category_summary loop finished; waiting...");
|
||||
thread::sleep(Duration::from_secs(1));
|
||||
}
|
||||
}
|
||||
|
||||
// NB: No failures at this level, because we don't want to prevent progress on a stuck feed.
|
||||
async fn single_job_loop_iteration(job_state: &Arc<JobState>) {
|
||||
let statuses = vec![AwaitableJobStatus::ReadyWaiting, AwaitableJobStatus::RetryablyFailed];
|
||||
for status in statuses {
|
||||
rendition_jobs_of_status(status, job_state).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn rendition_jobs_of_status(status: AwaitableJobStatus, job_state: &Arc<JobState>) {
|
||||
const BATCH_SIZE : i64 = 10;
|
||||
|
||||
let mut last_id = 0;
|
||||
let mut failure_count = 0;
|
||||
|
||||
loop {
|
||||
// NB: Protect sqlite from contention.
|
||||
thread::sleep(Duration::from_millis(500));
|
||||
|
||||
debug!("web_rendition querying {:?} targets from id > {} ...", &status, last_id);
|
||||
|
||||
let query_result = list_news_story_productions_awaiting_llm_rendition(
|
||||
status, last_id, BATCH_SIZE, &job_state.sqlite_pool).await;
|
||||
|
||||
let targets = match query_result {
|
||||
Ok(targets) => targets,
|
||||
Err(err) => {
|
||||
error!("failure querying batch: {:?}", err);
|
||||
|
||||
failure_count += 1;
|
||||
|
||||
// NB: Don't starve progress.
|
||||
if failure_count > 2 {
|
||||
failure_count = 0;
|
||||
last_id += 1;
|
||||
} else if failure_count > 3 {
|
||||
return;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
failure_count = 0;
|
||||
|
||||
if targets.is_empty() {
|
||||
return; // Done with batches.
|
||||
}
|
||||
|
||||
for target in targets {
|
||||
debug!("Rendition for target: {:?}", target.original_news_canonical_url);
|
||||
|
||||
let result = process_single_item(&target, job_state).await;
|
||||
if let Err(err) = result {
|
||||
error!("Error processing target: {:?}", err);
|
||||
}
|
||||
|
||||
last_id = target.id;
|
||||
|
||||
thread::sleep(Duration::from_secs(5)); // NB: We probably (?) need to throttle API calls to OpenAI
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
pub mod gpt_prompts;
|
||||
pub mod main_loop;
|
||||
pub mod process_single_item;
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
use async_openai::Client;
|
||||
use async_openai::error::OpenAIError;
|
||||
use async_openai::types::CreateCompletionRequestArgs;
|
||||
use crate::persistence::load_scraped_result::load_scraped_result;
|
||||
use crate::persistence::rendition_data::RenditionData;
|
||||
use crate::persistence::save_directory::SaveDirectory;
|
||||
use crate::shared_state::job_state::JobState;
|
||||
use crate::workers::news_stories::phase_2::news_story_llm_rendition::gpt_prompts::news_article_prompt::NewsArticlePrompt;
|
||||
use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus;
|
||||
use errors::{anyhow, AnyhowResult};
|
||||
use log::{error, info, warn};
|
||||
use sqlite_queries::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem;
|
||||
use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_llm_rendition_status::{Args, update_news_story_llm_rendition_status};
|
||||
use std::sync::Arc;
|
||||
use web_scrapers::payloads::web_scraping_result::ScrapedWebArticle;
|
||||
|
||||
pub async fn process_single_item(target: &NewsStoryProductionItem, job_state: &Arc<JobState>) -> AnyhowResult<()> {
|
||||
|
||||
let scraping_result = load_scraped_result(
|
||||
&target.original_news_canonical_url,
|
||||
&job_state.save_directory).await?;
|
||||
|
||||
let prompt = NewsArticlePrompt::new(&scraping_result.paragraphs);
|
||||
let prompt = prompt.make_prompt();
|
||||
|
||||
let rendition_result = call_openai_gpt(
|
||||
&target.original_news_canonical_url,
|
||||
&prompt,
|
||||
&job_state.openai_client
|
||||
).await;
|
||||
|
||||
let llm_rendition_attempts = target.llm_rendition_attempts + 1;
|
||||
|
||||
let rendition_data = match rendition_result {
|
||||
Err(err) => {
|
||||
error!("Error using GPT: {:?}", err);
|
||||
|
||||
match &err {
|
||||
WrappedOpenAiError::Other(_) => {}
|
||||
WrappedOpenAiError::OpenAiError(inner) => {
|
||||
match inner {
|
||||
OpenAIError::Reqwest(e) => { error!("Request error: {:?}", e) }
|
||||
OpenAIError::ApiError(e) => { error!("API error: {:?}", e) }
|
||||
OpenAIError::JSONDeserialize(e) => { error!("JSON error: {:?}", e) }
|
||||
OpenAIError::FileSaveError(e) => { error!("File save error: {:?}", e) }
|
||||
OpenAIError::FileReadError(e) => { error!("File read error: {:?}", e) }
|
||||
OpenAIError::StreamError(e) => { error!("Stream error: {:?}", e) }
|
||||
OpenAIError::InvalidArgument(e) => { error!("Invalid args error: {:?}", e) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
update_news_story_llm_rendition_status(Args {
|
||||
news_story_token: &target.news_story_token,
|
||||
is_successful: false,
|
||||
llm_rendition_attempts,
|
||||
sqlite_pool: &job_state.sqlite_pool,
|
||||
}).await?; // NB: If these queries fail, we could get stuck in retry hell.
|
||||
|
||||
return Err(anyhow!("OpenAI Rendition Error: {:?}", err));
|
||||
}
|
||||
Ok(rendition_data) => rendition_data,
|
||||
};
|
||||
|
||||
update_news_story_llm_rendition_status(Args {
|
||||
news_story_token: &target.news_story_token,
|
||||
is_successful: true,
|
||||
llm_rendition_attempts,
|
||||
sqlite_pool: &job_state.sqlite_pool,
|
||||
}).await?; // NB: If these queries fail, we could get stuck in retry hell.
|
||||
|
||||
// update_web_rendition_target(Args {
|
||||
// canonical_url: &target.original_news_canonical_url,
|
||||
// rendition_status: RenditionStatus::Success,
|
||||
// rendition_attempts: target.llm_rendition_attempts + 1,
|
||||
// sqlite_pool: &job_state.sqlite_pool,
|
||||
// }).await?; // NB: If these queries fail, we could get stuck.
|
||||
|
||||
{
|
||||
let yaml_filename = job_state.save_directory
|
||||
.rendition_file_for_webpage_url(&target.original_news_canonical_url)?;
|
||||
rendition_data.write_to_yaml_file(yaml_filename)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum WrappedOpenAiError {
|
||||
OpenAiError(OpenAIError),
|
||||
Other(String),
|
||||
}
|
||||
|
||||
async fn call_openai_gpt(url: &str, prompt: &str, openai_client: &Arc<Client>) -> Result<RenditionData, WrappedOpenAiError> {
|
||||
warn!("Building OpenAI Request...");
|
||||
let request = CreateCompletionRequestArgs::default()
|
||||
.model("text-davinci-003")
|
||||
.prompt(prompt)
|
||||
.max_tokens(2500_u16)
|
||||
.build()
|
||||
.map_err(|err| {
|
||||
error!("Error building request: {:?}", err);
|
||||
WrappedOpenAiError::Other(format!("error: {:?}", err))
|
||||
})?;
|
||||
|
||||
warn!("Making OpenAI Request...");
|
||||
let response = openai_client
|
||||
.completions()
|
||||
.create(request)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
error!("OpenAI Error: {:?}", err);
|
||||
WrappedOpenAiError::OpenAiError(err)
|
||||
})?;
|
||||
|
||||
info!("Open AI response: {:?}", response);
|
||||
|
||||
let rendition_text = response.choices.get(0)
|
||||
.map(|option| option.text.clone())
|
||||
.unwrap_or("".to_string());
|
||||
|
||||
Ok(RenditionData {
|
||||
original_content_url: url.to_string(),
|
||||
original_prompt: prompt.to_string(),
|
||||
response: rendition_text,
|
||||
})
|
||||
}
|
||||
|
||||
//fn next_status(rendition_attempts: i64) -> RenditionStatus {
|
||||
// if rendition_attempts >= 2 {
|
||||
// RenditionStatus::PermanentlyFailed
|
||||
// } else {
|
||||
// RenditionStatus::Failed
|
||||
// }
|
||||
//}
|
||||
+1
@@ -0,0 +1 @@
|
||||
pub mod news_article_prompt;
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
|
||||
// TODO: We'll add more types of prompt templates in the future.
|
||||
|
||||
pub struct NewsArticlePrompt {
|
||||
// TODO: Add other details (title, etc.)
|
||||
paragraphs: Vec<String>,
|
||||
}
|
||||
|
||||
impl NewsArticlePrompt {
|
||||
|
||||
pub fn new(paragraphs: &Vec<String>) -> Self {
|
||||
Self {
|
||||
paragraphs: paragraphs.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: This will need a lot of work and experimentation.
|
||||
pub fn make_prompt(&self) -> String {
|
||||
let start = "The following is a news article:\n\n";
|
||||
|
||||
let end = "\n\nPlease reword the above article as if you were a newscaster presenting the news. Spell out any abbreviations and numbers";
|
||||
|
||||
format!("{}{}{}", start, &self.paragraphs.join("\n\n"), end)
|
||||
}
|
||||
}
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
use crate::shared_state::job_state::JobState;
|
||||
use crate::workers::news_stories::phase_2::news_story_llm_rendition::process_single_item::process_single_item;
|
||||
use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus;
|
||||
use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
|
||||
use log::{debug, error, info};
|
||||
use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_llm_rendition::list_news_story_productions_awaiting_llm_rendition;
|
||||
use std::sync::Arc;
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
/// Project scraped, summarize the title.
|
||||
pub async fn news_story_llm_title_summary_main_loop(job_state: Arc<JobState>) {
|
||||
loop {
|
||||
debug!("news_story_llm_title_summary_main_loop main loop");
|
||||
|
||||
single_job_loop_iteration(&job_state).await;
|
||||
|
||||
debug!("llm_title_summary loop finished; waiting...");
|
||||
thread::sleep(Duration::from_secs(1));
|
||||
}
|
||||
}
|
||||
|
||||
// NB: No failures at this level, because we don't want to prevent progress on a stuck feed.
|
||||
async fn single_job_loop_iteration(job_state: &Arc<JobState>) {
|
||||
let statuses = vec![AwaitableJobStatus::ReadyWaiting, AwaitableJobStatus::RetryablyFailed];
|
||||
for status in statuses {
|
||||
rendition_jobs_of_status(status, job_state).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn rendition_jobs_of_status(status: AwaitableJobStatus, job_state: &Arc<JobState>) {
|
||||
const BATCH_SIZE : i64 = 10;
|
||||
|
||||
let mut last_id = 0;
|
||||
let mut failure_count = 0;
|
||||
|
||||
loop {
|
||||
// NB: Protect sqlite from contention.
|
||||
thread::sleep(Duration::from_millis(500));
|
||||
|
||||
debug!("web_rendition querying {:?} targets from id > {} ...", &status, last_id);
|
||||
|
||||
let query_result = list_news_story_productions_awaiting_llm_rendition(
|
||||
status, last_id, BATCH_SIZE, &job_state.sqlite_pool).await;
|
||||
|
||||
let targets = match query_result {
|
||||
Ok(targets) => targets,
|
||||
Err(err) => {
|
||||
error!("failure querying batch: {:?}", err);
|
||||
|
||||
failure_count += 1;
|
||||
|
||||
// NB: Don't starve progress.
|
||||
if failure_count > 2 {
|
||||
failure_count = 0;
|
||||
last_id += 1;
|
||||
} else if failure_count > 3 {
|
||||
return;
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
failure_count = 0;
|
||||
|
||||
if targets.is_empty() {
|
||||
return; // Done with batches.
|
||||
}
|
||||
|
||||
for target in targets {
|
||||
debug!("Rendition for target: {:?}", target.original_news_canonical_url);
|
||||
|
||||
let result = process_single_item(&target, job_state).await;
|
||||
if let Err(err) = result {
|
||||
error!("Error processing target: {:?}", err);
|
||||
}
|
||||
|
||||
last_id = target.id;
|
||||
|
||||
thread::sleep(Duration::from_secs(5)); // NB: We probably (?) need to throttle API calls to OpenAI
|
||||
}
|
||||
}
|
||||
}
|
||||
+3
@@ -0,0 +1,3 @@
|
||||
pub mod gpt_prompts;
|
||||
pub mod main_loop;
|
||||
pub mod process_single_item;
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
use async_openai::Client;
|
||||
use async_openai::error::OpenAIError;
|
||||
use async_openai::types::CreateCompletionRequestArgs;
|
||||
use crate::persistence::load_scraped_result::load_scraped_result;
|
||||
use crate::persistence::rendition_data::RenditionData;
|
||||
use crate::persistence::save_directory::SaveDirectory;
|
||||
use crate::shared_state::job_state::JobState;
|
||||
use crate::workers::news_stories::phase_2::news_story_llm_rendition::gpt_prompts::news_article_prompt::NewsArticlePrompt;
|
||||
use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus;
|
||||
use errors::{anyhow, AnyhowResult};
|
||||
use log::{error, info, warn};
|
||||
use sqlite_queries::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem;
|
||||
use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_llm_rendition_status::{Args, update_news_story_llm_rendition_status};
|
||||
use std::sync::Arc;
|
||||
use web_scrapers::payloads::web_scraping_result::ScrapedWebArticle;
|
||||
|
||||
pub async fn process_single_item(target: &NewsStoryProductionItem, job_state: &Arc<JobState>) -> AnyhowResult<()> {
|
||||
|
||||
let scraping_result = load_scraped_result(
|
||||
&target.original_news_canonical_url,
|
||||
&job_state.save_directory).await?;
|
||||
|
||||
let prompt = NewsArticlePrompt::new(&scraping_result.paragraphs);
|
||||
let prompt = prompt.make_prompt();
|
||||
|
||||
let rendition_result = call_openai_gpt(
|
||||
&target.original_news_canonical_url,
|
||||
&prompt,
|
||||
&job_state.openai_client
|
||||
).await;
|
||||
|
||||
let llm_rendition_attempts = target.llm_rendition_attempts + 1;
|
||||
|
||||
let rendition_data = match rendition_result {
|
||||
Err(err) => {
|
||||
error!("Error using GPT: {:?}", err);
|
||||
|
||||
match &err {
|
||||
WrappedOpenAiError::Other(_) => {}
|
||||
WrappedOpenAiError::OpenAiError(inner) => {
|
||||
match inner {
|
||||
OpenAIError::Reqwest(e) => { error!("Request error: {:?}", e) }
|
||||
OpenAIError::ApiError(e) => { error!("API error: {:?}", e) }
|
||||
OpenAIError::JSONDeserialize(e) => { error!("JSON error: {:?}", e) }
|
||||
OpenAIError::FileSaveError(e) => { error!("File save error: {:?}", e) }
|
||||
OpenAIError::FileReadError(e) => { error!("File read error: {:?}", e) }
|
||||
OpenAIError::StreamError(e) => { error!("Stream error: {:?}", e) }
|
||||
OpenAIError::InvalidArgument(e) => { error!("Invalid args error: {:?}", e) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
update_news_story_llm_rendition_status(Args {
|
||||
news_story_token: &target.news_story_token,
|
||||
is_successful: false,
|
||||
llm_rendition_attempts,
|
||||
sqlite_pool: &job_state.sqlite_pool,
|
||||
}).await?; // NB: If these queries fail, we could get stuck in retry hell.
|
||||
|
||||
return Err(anyhow!("OpenAI Rendition Error: {:?}", err));
|
||||
}
|
||||
Ok(rendition_data) => rendition_data,
|
||||
};
|
||||
|
||||
update_news_story_llm_rendition_status(Args {
|
||||
news_story_token: &target.news_story_token,
|
||||
is_successful: true,
|
||||
llm_rendition_attempts,
|
||||
sqlite_pool: &job_state.sqlite_pool,
|
||||
}).await?; // NB: If these queries fail, we could get stuck in retry hell.
|
||||
|
||||
// update_web_rendition_target(Args {
|
||||
// canonical_url: &target.original_news_canonical_url,
|
||||
// rendition_status: RenditionStatus::Success,
|
||||
// rendition_attempts: target.llm_rendition_attempts + 1,
|
||||
// sqlite_pool: &job_state.sqlite_pool,
|
||||
// }).await?; // NB: If these queries fail, we could get stuck.
|
||||
|
||||
{
|
||||
let yaml_filename = job_state.save_directory
|
||||
.rendition_file_for_webpage_url(&target.original_news_canonical_url)?;
|
||||
rendition_data.write_to_yaml_file(yaml_filename)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum WrappedOpenAiError {
|
||||
OpenAiError(OpenAIError),
|
||||
Other(String),
|
||||
}
|
||||
|
||||
async fn call_openai_gpt(url: &str, prompt: &str, openai_client: &Arc<Client>) -> Result<RenditionData, WrappedOpenAiError> {
|
||||
warn!("Building OpenAI Request...");
|
||||
let request = CreateCompletionRequestArgs::default()
|
||||
.model("text-davinci-003")
|
||||
.prompt(prompt)
|
||||
.max_tokens(2500_u16)
|
||||
.build()
|
||||
.map_err(|err| {
|
||||
error!("Error building request: {:?}", err);
|
||||
WrappedOpenAiError::Other(format!("error: {:?}", err))
|
||||
})?;
|
||||
|
||||
warn!("Making OpenAI Request...");
|
||||
let response = openai_client
|
||||
.completions()
|
||||
.create(request)
|
||||
.await
|
||||
.map_err(|err| {
|
||||
error!("OpenAI Error: {:?}", err);
|
||||
WrappedOpenAiError::OpenAiError(err)
|
||||
})?;
|
||||
|
||||
info!("Open AI response: {:?}", response);
|
||||
|
||||
let rendition_text = response.choices.get(0)
|
||||
.map(|option| option.text.clone())
|
||||
.unwrap_or("".to_string());
|
||||
|
||||
Ok(RenditionData {
|
||||
original_content_url: url.to_string(),
|
||||
original_prompt: prompt.to_string(),
|
||||
response: rendition_text,
|
||||
})
|
||||
}
|
||||
|
||||
//fn next_status(rendition_attempts: i64) -> RenditionStatus {
|
||||
// if rendition_attempts >= 2 {
|
||||
// RenditionStatus::PermanentlyFailed
|
||||
// } else {
|
||||
// RenditionStatus::Failed
|
||||
// }
|
||||
//}
|
||||
Reference in new issue
Block a user