The LLM Framework Ecosystem

Key primitives:

PromptTemplate → prompt template with variables
ChatModel → wrapper for LLM API (OpenAI, Anthropic, ...)
OutputParser → output parsing: str/JSON/Pydantic
Retriever → search abstraction (vector, BM25, hybrid)
Tool → called tool with description for LLM
Memory → conversation context storage
Chain → an outdated way to connect components
LCEL pipe → prompt | model | parser ← modern way

# Simple RAG chain
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt_template
    | chat_model
    | StrOutputParser()
)

# Everything works automatically:
chain.invoke(query) # synchronously
chain.stream(query) # token streaming
chain.batch([q1, q2, q3]) # in parallel
await chain.ainvoke(query) #async

✗ Magic prompts
  Default prompts inside Chain classes
  → unpredictable, impossible to control

✗ Hidden LLM challenges
  RetrievalQA does extra. calls unnoticed
  → unexpected costs, hidden latency

✗ Versioning
  Frequent breaking changes between minor versions
  → pin version, test updates carefully

✓ Solution: LCEL + explicit prompts + tracing everything

Agent graph: ReAct with human-in-the-loop

      [START]
         ↓
      [agent] ← LLM decides: call the tool or give a response
       ↙ ↘
[tools] [END]
   ↓
[agent] ← tool result → next solution

Key idea: cycles are possible - the agent can spin
as much as needed until it solves the problem or hits the limit

from typing import Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages = reducer: append, not replace
    retrieved_docs: list[str]
    iteration_count: int
    final_answer: str | None

# each node receives state and returns patch:
def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools" # → tools node
    elif state["iteration_count"] > 5:
        return "force_end" # → force stop
    else:
        return "end" # → final response

graph.add_conditional_edges("agent", should_continue,
    {"tools": "tools", "end": END, "force_end": END})

checkpointer = PostgresSaver(conn)
graph = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"] ← pause here
)

# 1. Run before interrupt
graph.invoke(input, config={"thread_id": "t1"})
# 2. Show the person a pending action
state = graph.get_state(config)
# 3. The person approves → continue
graph.invoke(None, config) # resume

LlamaIndex abstractions:

Document → raw document + metadata
Node → document chunk (atomic index unit)
Index → search structure (VectorStore, Summary, KG...)
Retriever → strategy for retrieving from Index
NodePostprocessor → reranker, filters, metadata enrichment
QueryEngine → retriever + LLM + response synthesis
ChatEngine → QueryEngine + dialogue history
Pipeline → IngestionPipeline for indexing

VectorStoreIndex standard RAG for embeddings
SummaryIndex summarization of the entire corpus sequentially
KeywordTableIndex keyword extraction → keyword search
KnowledgeGraph entity graph → search by relationships
DocumentSummary summary per doc → document selection → details
TreeIndex hierarchical tree of summations

Multi-Index: different indexes for different types of questions

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, overlap=64),
        TitleExtractor(), # LLM metadata
        QuestionsAnsweredExtractor(), # hypothetical questions
        OpenAIEmbedding(),
    ],
    vector_store=chroma_store,
    cache=IngestionCache(), # do not reindex old
)

CrewAI abstractions:

Agent = role + backstory + goal + tools + llm
Task = description + expected_output + agent
Crew = [agents] + [tasks] + process + memory

Process types:
  sequential agent 1 → agent 2 → agent 3
  hierarchical manager LLM → delegates tasks to agents
                             ↑ better quality control

Crew example for WB seller analytics:
researcher → collect data on competitors
analyst → build BCG matrix
writer → write a report in PDF

ConversableAgent base agent with LLM + human_input_mode
AssistantAgent LLM agent without human input
UserProxyAgent executes code, optionally requests a person
GroupChat multiple agents + next selection manager
GroupChatManager LLM decides who speaks next

Pattern: Coder + Reviewer + Executor
Coder writes → Reviewer criticizes → Executor launches

agent = Agent(
    model=Claude(id="claude-sonnet-4"),
    tools=[DuckDuckGoSearch(), PythonTool()],
    memory=AgentMemory(db=SqliteMemoryDb()),
    knowledge=PDFKnowledgeBase(path="docs/"),
    instructions=["Answer in Russian", "Use numbers"],
    show_tool_calls=True,
    markdown=True,
)
agent.print_response("WB 2025 Market Analysis")

What RAGAS evaluates and how:

Faithfulness
  answer → atomic statements → NLI vs context
  = proportion of statements supported by context

Answer Relevance
  answer → LLM generates N questions → embed → cos(questions, query)
  = to what extent the answer falls within the topic of the question

Context Precision
  Each chunk → LLM : “Is LLM useful for an answer?” → precision@k
  = share of useful chunks among retrieved

Context Recall [requires GT]
  GT → atomic statements → NLI vs context
  = proportion of GT facts present in the context

@pytest.mark.parametrize("case", test_cases)
def test_rag_quality(case):
    actual_output = rag_pipeline(case.input)

    assert_test(
        test_case=LLMTestCase(
            input=case.input,
            actual_output=actual_output,
            retrieval_context=case.context
        ),
        metrics=[
            FaithfulnessMetric(threshold=0.8),
            AnswerRelevancyMetric(threshold=0.7),
            HallucinationMetric(threshold=0.1),
        ]
    )

Eval(
    name="RAG pipeline v2 vs v3",
    data=dataset, # list {input, expected}
    task=rag_pipeline, # function → output
    scores=[ # metrics
        Faithfulness,
        AnswerRelevancy,
        NumTokens, # cost tracking
        Latency
    ],
    experiment_name="v3-with-rerank"
)

# One-time use:
human_labels = 150 labeled examples
synthetic_data = LLM.generate_training_data(docs)
classifier = finetune(small_lm, human_labels + synthetic)

# Next - cheap and fast:
scores = classifier.score(10_000_examples)
# vs LLM-judge: 100x cheaper, comparable quality

What Langfuse can do:

Traces & Spans pipeline visualization, waterfall latency
Scores eval-assessments attached to trace/span
Datasets golden datasets + experiments (A/B eval)
Prompt Management versioning prompts + fetch at runtime
Cost Tracking tokens and $ by model, user, feature
Users & Sessions grouping traces by user
Evaluators LLM-as-judge directly in the interface

Integration: 2 lines of code
from langfuse.openai import openai # drop-in replacement

Embedding visualization
  UMAP/TSNE projections: see query clusters
  → find anomalies and OOD queries visually

Retrieval analysis
  scatter plot: query embeddings vs doc embeddings
  → see where retrieval misses

Drift detection
  comparison of distributions between periods
  → automatically finds data drift

OpenInference
  open tracing standard LLM (like OTEL)

client = OpenAI(
    base_url="https://oai.helicone.ai/v1",
    # ← instead of api.openai.com
    default_headers={
        "Helicone-Auth": f"Bearer {HELICONE_KEY}",
        "Helicone-Cache-Enabled": "true", # cache
        "Helicone-User-Id": user_id, # per-user
    }
)

gen_ai.system = "anthropic"
gen_ai.request.model = "cloud-sonnet-4"
gen_ai.usage.input_tokens = 1240
gen_ai.usage.output_tokens = 312
gen_ai.prompt = [messages]
gen_ai.completion = [response]
# → goes to Langfuse / Jaeger / Grafana Tempo

Traditional approach:
  the prompt was written manually → works on GPT-4 → the model was changed → the prompt is broken

DSPy approach:
  signature = "question → answer" # what we do
  metric = lambda pred, gt: f1(pred, gt) # how to measure
  optimizer.compile(program, metric, trainset)
  # DSPy generates prompts itself + few-shot examples
  # re-optimizes when changing models in minutes

Key primitives:
Signature InputField + OutputField = declarative specification
Module dspy.Predict / dspy.ChainOfThought / dspy.ReAct
Optimizer BootstrapFewShot/MIPRO/BayesianSignatureOptimizer
Metric function (prediction, ground_truth) → float

dspy.Predict
  direct call by signature, without CoT

dspy.ChainOfThought
  adds a reasoning field → response
  # prompt with CoT is generated automatically

dspy.ReAct
  agent with tools based on the ReAct pattern

dspy.Retrieve
  search in the connected knowledge base

dspy.MultiChainComparison
  several reasoning paths → the best

loss = metric(output, gt) # "how bad"
gradient = LLM( # "why is it bad"
    f"Output: {output}
Loss: {loss}
What's wrong with the prompt?"
)
new_prompt = LLM( # "fix"
    f"Old prompt: {prompt}
Gradient: {gradient}
Improved prompt:"
)

What are you building?
 ├── RAG with data → LlamaIndex (indices) + LangChain (orchestration)
 ├── Stateful agent with cycles → LangGraph
 ├── Multi-agent business process → CrewAI (fast) / LangGraph (reliable)
 ├── Research/agent dialogue → AutoGen
 ├── Simple agent, control is important → Agno / pure SDK
 └── Prompts to optimize auto → DSPy

What are you assessing?
 ├── RAG (without GT) → RAGAS
 ├── LLM tests in pytest → DeepEval
 ├── Experiments / A/B prompts → Braintrust
 └── Small team, all at once → Langfuse (datasets + experiments)

What are you monitoring?
 ├── Traces + prompts + cost → Langfuse (self-hosted)
 ├── Embeddings drift + retrieval → Phoenix Arize
 └── Zero instrumentation, tomorrow → Helicone

✗ One or two LLM challenges
  → direct SDK (anthropic, openai) is better

✗ The team does not know the framework
  → time to study > time to write yourself

✗ Specific logic in every step
  → abstractions get in the way, you write workarounds

✓ The framework is justified when:
  → 10+ integrations are needed (providers, vector databases)
  → complex stateful graph (LangGraph)
  → the team already knows him

Minimalistic (small team):
  SDK + Instructor + Langfuse

RAG product:
  LlamaIndex + LangGraph (agent) + RAGAS + Langfuse

Multi-agent enterprise:
  LangGraph + LangChain tools + Braintrust + Arize

Research/Experiments:
  DSPy + Langfuse + DeepEval

Main principle:
  start with the minimum → add frameworks
  only when the pain without them is obvious