writer.py

from email import header
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
import phoenix as px

px.launch_app()
endpoint = "http://neutral-sole-glad.ngrok-free.app/v1/traces"
resource = Resource(attributes={})
tracer_provider = trace_sdk.TracerProvider(resource=resource)
span_otlp_exporter = OTLPSpanExporter(
    endpoint=endpoint, headers={"ngrok-skip-browser-warning": "True"}
)
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, styled in markdown"
    )


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