The LLM Framework Ecosystem

# 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

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

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
)

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")

@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

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 = "claude-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

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:"
)

✗ 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