code.py

"""Parallelization — run independent LLM calls at the same time, then merge.

Three analyses of the same document don't depend on each other, so running
them in sequence just adds up their latencies for no reason. RunnableParallel
fires them concurrently and hands back a dict of results.

The catch: only fan out work that is genuinely independent. If step B needs
step A's output, you have a chain, not a fan-out.
"""

import time

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel

from resources.agent import llm
from resources.helper import show_response

REVIEW = """
The battery lasts about six hours under load, well short of the twelve the box
claims. Build quality is genuinely excellent - the hinge is solid and the
chassis doesn't flex. Support took nine days to answer a simple question. At
$1,800 it's hard to recommend over the competition.
"""

# Three independent lenses on the same text. None of them needs another's
# output, which is exactly what makes the fan-out legal.
BRANCHES = {
    "sentiment": "Rate the sentiment of this review as positive, negative, or "
    "mixed, and justify it in one sentence.\n\n{text}",
    "complaints": "List only the concrete complaints in this review, as bullet "
    "points. No commentary.\n\n{text}",
    "praise": "List only what this review praises, as bullet points. No "
    "commentary.\n\n{text}",
}

# RunnableParallel takes a dict of runnables and executes them concurrently,
# returning a dict with the same keys. This is the whole pattern.
fan_out = RunnableParallel(
    {
        name: ChatPromptTemplate.from_template(template) | llm
        for name, template in BRANCHES.items()
    }
)

# The merge step. Fan-out is only half the pattern - something has to reduce
# the branch results back into one answer.
merge_prompt = ChatPromptTemplate.from_template(
    "Write a two-sentence summary for a product manager, using these analyses.\n\n"
    "Sentiment: {sentiment}\n\nComplaints: {complaints}\n\nPraise: {praise}"
)

if __name__ == "__main__":
    start = time.perf_counter()
    branches = fan_out.invoke({"text": REVIEW})
    parallel_seconds = time.perf_counter() - start

    for name, message in branches.items():
        print(f"\n{'=' * 60}\nBRANCH: {name}\n{'=' * 60}")
        show_response(message)

    # Merge the three branch outputs into the final deliverable.
    print(f"\n{'=' * 60}\nMERGED\n{'=' * 60}")
    show_response(
        llm.invoke(
            merge_prompt.format(**{k: v.content for k, v in branches.items()})
        )
    )

    # The payoff, measured rather than asserted: three calls took roughly as
    # long as the slowest one, not the sum of all three.
    print(f"\n[timing] 3 branches finished in {parallel_seconds:.1f}s total")