Spaces:
Sleeping
Sleeping
from langgraph.graph import START, END, StateGraph | |
from blogGenerator.state.state import State | |
from blogGenerator.nodes.route_deciding_node import RouteDecidingNode | |
from blogGenerator.nodes.yt_transcript_node import YTTranscriptNode | |
from blogGenerator.nodes.content_generator_node import ContentGeneratorNode | |
class GraphBuilder: | |
def __init__(self, model): | |
self.llm = model | |
self.graph_builder = StateGraph(State) | |
def blog_generator_build_graph(self): | |
""" | |
Build a blog generation graph | |
""" | |
# print("Build a blog generation graph.......") | |
try: | |
self.route_deciding_node = RouteDecidingNode(self.llm) | |
self.yt_transcript_node = YTTranscriptNode() | |
self.content_generator_node = ContentGeneratorNode(self.llm) | |
self.graph_builder.add_node("Router", self.route_deciding_node.process) | |
self.graph_builder.add_node( | |
"YT Transcript", self.yt_transcript_node.process | |
) | |
self.graph_builder.add_node( | |
"Generate Blog", self.content_generator_node.process | |
) | |
self.graph_builder.add_edge(START, "Router") | |
self.graph_builder.add_conditional_edges( | |
"Router", | |
self.route_deciding_node.route_decision, | |
{"youtube": "YT Transcript", "topic": "Generate Blog"}, | |
) | |
self.graph_builder.add_edge("YT Transcript", "Generate Blog") | |
self.graph_builder.add_edge("Generate Blog", END) | |
except Exception as e: | |
print(f"Error : {e}") | |
def setup_graph(self): | |
self.blog_generator_build_graph() | |
return self.graph_builder.compile() | |