-
Notifications
You must be signed in to change notification settings - Fork 4
/
bloggpt.py
190 lines (159 loc) · 5.81 KB
/
bloggpt.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
# Purpose: Run the bloggpt agent
import logging
import os
import sys
import traceback
from pprint import pprint
from typing import List, Optional
import streamlit as st
from ansi2html import Ansi2HTMLConverter
from dotenv import load_dotenv
from langchain.agents import AgentType, Tool, initialize_agent
from langchain.chat_models import ChatOpenAI
from prompts.prompts import (
BLOG_SECTION_AGENT_SYSTEM_PROMPT,
OUTLINE_PROMPT,
REWRITE_PROMPT,
TOPIC_PROMPT,
)
from utils.logging_utils import StreamlitPrint
from utils.main_utils import (
bprint,
combine_drafts,
generate_final_blog,
gprint,
rprint,
split_outline_prompt,
)
from utils.web_utils import search_and_summarize_web_url
sys.stdout = StreamlitPrint()
# Load environment variables from .env file
load_dotenv(".")
# Use the environment variables
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
tools = [search_and_summarize_web_url]
chat_model = ChatOpenAI(
model_name="gpt-4-0613",
temperature=0,
openai_api_key=OPENAI_API_KEY,
)
# https://github.com/hwchase17/langchain/issues/6025
blog_agent = initialize_agent(
tools,
chat_model,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
max_iterations=4,
)
def get_topic_context(topic: str) -> Optional[str]:
"""
Get the search context for a given topic from the web.
Args:
topic: A string representing the topic.
Returns:
Relevant information retrieved from the web about the topic.
If an error occurs during the search, returns None.
"""
with st.expander(f"Search Results For: {topic}"):
try:
context = search_and_summarize_web_url(topic)
st.write(context)
except Exception as e:
st.error(f"Error occurred while searching for {topic}: {e}")
st.error(traceback.format_exc())
raise e
st.divider()
return context
def generate_blog_section(
header: str, blog_section: str, context: str
) -> Optional[str]:
"""
Generate a blog section for a given header, blog_section and context.
Args:
header: A string representing the header.
blog_section: A string representing the blog section.
context: A string representing the context.
Returns:
A string representing the generated blog section.
If an error occurs during the generation, returns None.
"""
try:
with st.expander(f"Generating Blog Section: {header}"):
GENERATE_BLOG_SECTION_PROMPT = BLOG_SECTION_AGENT_SYSTEM_PROMPT.format(
TOPIC_PROMPT=TOPIC_PROMPT,
BLOG_SECTION_OUTLINE_PROMPT=blog_section,
CONTEXT=context,
)
logging.debug(GENERATE_BLOG_SECTION_PROMPT)
rprint(f"Generating Blog Section: {header}")
generated_blog = blog_agent.run(GENERATE_BLOG_SECTION_PROMPT)
except Exception as e:
st.error(f"Error occurred while generating blog section {header}: {e}")
st.error(traceback.format_exc())
generated_blog = None
return generated_blog
def save_blog_section(generated_blog: str, header: str, num_generated: int) -> None:
"""
Save a generated blog section.
Args:
generated_blog: A string representing the generated blog.
header: A string representing the header.
num_generated: An integer representing the number of generated sections.
Returns:
None. If an error occurs during the saving process, logs the error.
"""
try:
with st.expander(f"Blog Section: {header}"):
st.write(generated_blog)
with open(f"outputs/draft_{num_generated}.md", "w") as f:
f.write(generated_blog)
except Exception as e:
st.error(f"Error occurred while saving blog section {header}: {e}")
st.error(traceback.format_exc())
st.divider()
def combine_and_finalize_draft(generated_blogs: List[str]) -> None:
"""
Combine and finalize the draft.
Args:
headers: A list of strings representing the headers.
blog_sections: A list of strings representing the blog sections.
Returns:
None. If an error occurs during the process, logs the error.
"""
try:
with st.expander("Blog Draft"):
rprint("Combining the generated blog sections into one")
entire_draft = "\n\n".join(generated_blogs)
st.write(entire_draft)
# entire_draft = combine_drafts("outputs/")
with st.expander("Final Blog"):
rprint("Generating Final Blog")
final_blog = generate_final_blog(entire_draft, TOPIC_PROMPT, OPENAI_API_KEY)
gprint("Done!")
st.write(final_blog)
except Exception as e:
logging.error(f"Error occurred while finalizing blog: {e}")
st.error(traceback.format_exc())
@st.cache_data()
def run_bloggpt(topic_str: str, blog_outline: str) -> None:
"""
Run bloggpt to generate a blog from provided topic and blog outline.
topic_str: A string representing the topic.
blog_outline: A string representing the blog outline.
Returns:
None.
"""
topic = topic_str.split(":")[1].strip()
context = get_topic_context(topic)
headers, blog_sections = split_outline_prompt(blog_outline)
generated_blogs = []
for num_generated, (header, blog_section) in enumerate(zip(headers, blog_sections)):
generated_blog = generate_blog_section(header, blog_section, context)
if generated_blog is not None:
generated_blogs.append(generated_blog)
with st.expander(f"Blog Section: {header}"):
st.write(generated_blog)
# save_blog_section(generated_blog, header, num_generated)
combine_and_finalize_draft(generated_blogs)
if __name__ == "__main__":
run_bloggpt(TOPIC_PROMPT, OUTLINE_PROMPT)