content-engine/llm.py at main · OutRizz/content-engine · GitHub

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
"""LLM calls: generate post ideas (structured output)."""

from __future__ import annotations

import json

from langfuse.openai import openai

from models import PostIdea
from prompt_loader import load_prompt

POST_IDEAS_JSON_SCHEMA: dict = {
    "type": "object",
    "properties": {
        "post_ideas": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "source_links": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "One or more URLs of source articles (from the Sources list). Use several links when the idea combines insights from multiple articles.",
                    },
                    "source_insight": {
                        "type": "string",
                        "description": "Key idea from the source material — what the article(s) are about, the core insight you are building on",
                    },
                    "post_idea": {
                        "type": "string",
                        "description": "The actual post idea: headline or hook for the post (one sentence)",
                    },
                    "description": {
                        "type": "string",
                        "description": "Short description of what the post will be about",
                    },
                    "recommended_format": {
                        "type": "string",
                        "description": "Recommended format for the post (e.g. LinkedIn post, thread, short article, carousel)",
                    },
                    "how_to_use": {
                        "type": "string",
                        "description": "Explanation of how to use the idea in the post",
                    },
                },
                "required": [
                    "source_links",
                    "source_insight",
                    "post_idea",
                    "description",
                    "recommended_format",
                    "how_to_use",
                ],
                "additionalProperties": False,
            },
        }
    },
    "required": ["post_ideas"],
    "additionalProperties": False,
}


def generate_post_ideas(
    content: str, model: str, count: int = 10, sources_list: str = ""
) -> list[PostIdea]:
    """Generate post ideas from full parsing results; returns structured list."""
    system = load_prompt("post_ideas_system")
    user = load_prompt(
        "post_ideas_user",
        count=str(count),
        content=content,
        sources=sources_list,
    )

    response = openai.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        name="content-engine-post-ideas",
        metadata={"step": "post_ideas"},
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "post_ideas_response",
                "strict": True,
                "schema": POST_IDEAS_JSON_SCHEMA,
            },
        },
    )
    raw = response.choices[0].message.content or "{}"
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return []
    ideas = data.get("post_ideas", [])
    result: list[PostIdea] = []
    for item in ideas:
        if not isinstance(item, dict):
            continue
        try:
            raw_links = item.get("source_links", [])
            links = tuple(str(u) for u in raw_links if isinstance(u, str) and u.strip())
            result.append(
                PostIdea(
                    source_links=links if links else (),
                    source_insight=str(item.get("source_insight", "")),
                    post_idea=str(item.get("post_idea", "")),
                    description=str(item.get("description", "")),
                    recommended_format=str(item.get("recommended_format", "")),
                    how_to_use=str(item.get("how_to_use", "")),
                )
            )
        except (TypeError, ValueError):
            continue
    return result