import typing import dspy from dspy import Signature, InputField, Module, ChainOfThought, Predict import streamlit as st from typing import Optional from openinference.instrumentation.dspy import DSPyInstrumentor from opentelemetry import trace as trace_api from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace.export import SimpleSpanProcessor endpoint = "http://127.0.0.1:6006/v1/traces" resource = Resource(attributes={}) tracer_provider = trace_sdk.TracerProvider(resource=resource) span_otlp_exporter = OTLPSpanExporter(endpoint=endpoint) tracer_provider.add_span_processor( SimpleSpanProcessor(span_exporter=span_otlp_exporter) ) trace_api.set_tracer_provider(tracer_provider=tracer_provider) DSPyInstrumentor().instrument() base_lm = dspy.OpenAI( model="gpt-4", max_tokens=4096, api_key=st.secrets["OpenAI"], ) dspy.settings.configure(lm=base_lm) # refin_lm = dspy.OpenAI( # model="gpt-3.5-turbo", # max_tokens=4096, # api_key="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", # ) _template = """A few controversial things I believe about {Topic}: 1) {StrongOpinion1} I’m not saying {SubtleDistinction}. I’m just saying {ClarifyingStatement}. - 2) {StrongOpinion2} Contrary to popular belief, {ClicheAction} is a myth. - {FailureExample1} - {FailureExample2} - {FailureExample3} {ClicheAction} is a fast-track to {UndesirableOutcome}. - 3) {StrongOpinion3} “Most people” do what “most people” do. Which leads to… - {NegativeStat1} - {NegativeStat2} - {NegativeStat3} Clearly what “most people do” doesn’t work. So do something else. - 4) {StrongOpinion4} 99% of the time, this is a giant mistake. Instead, {Name} has a great framework for achieving {Outcome} without falling into {Trap}. - {Step1} - {Step2} - {Step3} - 5) {StrongOpinion5} Pursuing {Outcome} is a silly goal. Because {UnconventionalReason}. {DifferentOutcome} is a better pursuit in life. - {Benefit1} - {Benefit2} - {Benefit3}""" class GenerateLinkedinArticle(Signature): topic = InputField(desc="Title or brief description of content") template = InputField(prefix="template to follow") purpose = InputField() audience = InputField() tone_style = InputField() key_points = InputField() num_words = InputField() language = InputField() article = dspy.OutputField(desc=f"Linkedin article of {num_words} words") class LinkedinArticle(Module): def __init__(self): super().__init__() self.generate_article = ChainOfThought(GenerateLinkedinArticle) def forward( self, topic, template, purpose, audience, tone_style, key_points, num_words, language, ): article_prediction = self.generate_article( topic=topic, template=template, purpose=purpose, audience=audience, tone_style=tone_style, key_points=key_points, num_words=num_words, language=language, ) return article_prediction.article class Feedback(Module): def __init__(self): super().__init__() self.feedback = ChainOfThought("original_content, feedback -> refined_content") def forward(self, original_content, feedback): corrected_article = self.feedback( original_content=original_content, feedback=feedback ).refined_content return corrected_article class Evaluator(dspy.Signature): """You are a creative writing coach, evaluate this linkedin post""" linkedin_post = InputField(prefix="A linkedin post to evaluate") output = dspy.OutputField( prefix="A comprehensive evaluation of the input post, styled in markdown" ) class SelfEval(Module): def __init__(self): super().__init__() self.self_eval = ChainOfThought(Evaluator) def forward(self, linkedin_post): eval = self.self_eval(linkedin_post=linkedin_post).output return eval def write_article(user_inputs: dict, lm: Optional[dspy.dsp.LM] = base_lm) -> str: with dspy.context(lm=lm): article = LinkedinArticle() content = article.forward(**user_inputs) return content def incorporate_feedback( original_content, feedback, lm: Optional[dspy.dsp.LM] = base_lm ): with dspy.context(lm=lm): refined_content = Feedback().forward( original_content=original_content, feedback=feedback ) return refined_content def evaluate_post(content: str, lm: Optional[dspy.dsp.LM] = base_lm) -> str: with dspy.context(lm=lm): eval = SelfEval().forward(linkedin_post=content) return eval