LlamaIndex 面经

📖 "LLMs are trained on enormous bodies of data but they aren’t trained on your data. Retrieval-Augmented Generation (RAG) solves this problem by adding your data to the data LLMs already have access to." — Introduction to RAG | Developer Documentation

📖 "In RAG, your data is loaded and prepared for queries or “indexed”. User queries act on the index, which filters your data down to the most relevant context. This context and your query then go to the LLM along with a prompt, and the LLM provides a response." — Introduction to RAG | Developer Documentation

flowchart LR
  A[你的数据] -->|Loading| B[Index 索引]
  Q[用户 Query] --> B
  B -->|过滤出最相关 context| C[Context + Query + Prompt]
  C --> D[LLM]
  D --> E[Response 回答]

📖 "There are five key stages within RAG, which in turn will be a part of most larger applications you build." — Introduction to RAG | Developer Documentation

📖 "Evaluation: a critical step in any flow is checking how effective it is relative to other strategies, or when you make changes. Evaluation provides objective measures of how accurate, faithful and fast your responses to queries are." — Introduction to RAG | Developer Documentation

flowchart LR
  L[Loading] --> I[Indexing] --> S[Storing] --> Q[Querying] --> E[Evaluation]

📖 "A Document is a container around any data source - for instance, a PDF, an API output, or retrieve data from a database. A Node is the atomic unit of data in LlamaIndex and represents a “chunk” of a source Document. Nodes have metadata that relate them to the document they are in and to other nodes." — Introduction to RAG | Developer Documentation

📖 "Transformation input/outputs are Node objects (a Document is a subclass of a Node)." — global

from llama_index.core.schema import TextNode


node1 = TextNode(text="<text_chunk>", id_="<node_id>")
node2 = TextNode(text="<text_chunk>", id_="<node_id>")


index = VectorStoreIndex([node1, node2])

📖 "The easiest reader to use is our SimpleDirectoryReader, which creates documents out of every file in a given directory. It is built in to LlamaIndex and can read a variety of formats including Markdown, PDFs, Word documents, PowerPoint decks, images, audio and video." — global

📖 "Data connectors ingest data from different data sources and format the data into Document objects. A Document is a collection of data (currently text, and in future, images and audio) and metadata about that data." — global

from llama_index.core import SimpleDirectoryReader


documents = SimpleDirectoryReader("./data").load_data()

📖 "After the data is loaded, you then need to process and transform your data before putting it into a storage system. These transformations include chunking, extracting metadata, and embedding each chunk. This is necessary to make sure that the data can be retrieved, and used optimally by the LLM." — global

📖 "A key step to process your documents is to split them into “chunks”/Node objects. The key idea is to process your data into bite-sized pieces that can be retrieved / fed to the LLM." — global

from llama_index.core.node_parser import SentenceSplitter


text_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=10)

📖 "Indexes have a .from_documents() method which accepts an array of Document objects and will correctly parse and chunk them up. However, sometimes you will want greater control over how your documents are split up." — global

📖 "You can do this by either using our transformation modules (text splitters, metadata extractors, etc.) as standalone components, or compose them in our declarative Transformation Pipeline interface." — global

from llama_index.core import SimpleDirectoryReader
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import TokenTextSplitter


documents = SimpleDirectoryReader("./data").load_data()


pipeline = IngestionPipeline(transformations=[TokenTextSplitter(), ...])


nodes = pipeline.run(documents=documents)

📖 "A vector embedding, often just called an embedding, is a numerical representation of the semantics, or meaning of your text. Two pieces of text with similar meanings will have mathematically similar embeddings, even if the actual text is quite different." — Indexing | Developer Documentation

📖 "When you want to search your embeddings, your query is itself turned into a vector embedding, and then a mathematical operation is carried out by VectorStoreIndex to rank all the embeddings by how semantically similar they are to your query." — Indexing | Developer Documentation

from llama_index.core import VectorStoreIndex


index = VectorStoreIndex.from_documents(documents)

📖 "The number of embeddings it returns is known as k, so the parameter controlling how many embeddings to return is known as top_k. This whole type of search is often referred to as “top-k semantic retrieval” for this reason." — Indexing | Developer Documentation

📖 "Top-k retrieval is the simplest form of querying a vector index; you will learn about more complex and subtler strategies when you read the querying section." — Indexing | Developer Documentation

from llama_index.core.retrievers import VectorIndexRetriever


retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=10,
)

📖 "Retrieval is when you find and return the most relevant documents for your query from your Index." — build index

📖 "Response synthesis is when your query, your most-relevant data and your prompt are combined and sent to your LLM to return a response." — build index

flowchart LR
    Q[Query] --> R[Retrieval
top-k semantic]
    R --> P[Postprocessing
rerank/filter]
    P --> S[Response synthesis
send to LLM]
    S --> A[Response]

from llama_index.core import get_response_synthesizer
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor


query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer,
    node_postprocessors=[SimilarityPostprocessor(similarity_cutoff=0.7)],
)

📖 "compact: “compact” the prompt during each LLM call by stuffing as many Node text chunks that can fit within the maximum prompt size. If there are too many chunks to stuff in one prompt, “create and refine” an answer by going through multiple prompts." — build index

📖 "tree_summarize: Given a set of Node objects and the query, recursively construct a tree and return the root node as the response. Good for summarization purposes." — build index

query_engine = RetrieverQueryEngine.from_args(
    retriever, response_mode=response_mode
)

📖 "The simplest way to store your indexed data is to use the built-in .persist() method of every Index, which writes all the data to disk at the location specified. This works for any type of index." — rebuild storage context

📖 "The API calls to create the embeddings in a VectorStoreIndex can be expensive in terms of time and money, so you will want to store them to avoid having to constantly re-index things." — rebuild storage context

from llama_index.core import StorageContext, load_index_from_storage


# rebuild storage context
storage_context = StorageContext.from_defaults(persist_dir="<persist_dir>")


# load index
index = load_index_from_storage(storage_context)

import chromadb
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import StorageContext


# initialize client, setting path to save data
db = chromadb.PersistentClient(path="./chroma_db")


# create collection
chroma_collection = db.get_or_create_collection("quickstart")


# assign chroma as the vector_store to the context
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

📖 "In LlamaIndex terms, an Index is a data structure composed of Document objects, designed to enable querying by an LLM. Your Index is designed to be complementary to your querying strategy." — Indexing | Developer Documentation

📖 "A Summary Index is a simpler form of Index best suited to queries where, as the name suggests, you are trying to generate a summary of the text in your Documents. It simply stores all of the Documents and returns all of them to your query engine." — Indexing | Developer Documentation

from llama_index.core import VectorStoreIndex


index = VectorStoreIndex([])
for doc in documents:
    index.insert(doc)

flowchart LR
    A[接收任务] --> B[选择最合适的 tool]
    B --> C[执行该步]
    C --> D{任务完成?}
    D -- 是 --> E[返回结果给用户]
    D -- 否 --> B

📖 "In LlamaIndex, an agent is a semi-autonomous piece of software powered by an LLM that is given a task and executes a series of steps towards solving that task." — Building an agent | Developer Documentation

📖 "When each step is completed, the agent judges whether the task is now complete, in which case it returns a result to the user, or whether it needs to take another step, in which case it loops back to the start." — Building an agent | Developer Documentation

workflow = FunctionAgent(
    tools=[multiply, add],
    llm=llm,
    system_prompt="You are an agent that can perform basic mathematical operations using tools.",
)
response = await workflow.run(user_msg="What is 20+(2*4)?")
print(response)
# The result of (20 + (2 times 4)) is 28.

📖 "When deciding what tool to use, your agent will use the tool’s name, parameters, and docstring to determine what the tool does and whether it’s appropriate for the task at hand." — Building an agent | Developer Documentation

📖 "It will also use the type hints to determine the expected parameters and return type." — Building an agent | Developer Documentation

def multiply(a: float, b: float) -> float:
    """Multiply two numbers and returns the product"""
    return a * b

📖 "you can use our pre-built agentic workflows like FunctionAgent (a simple function/tool calling agent) or AgentWorkflow (an agent capable of managing multiple agents)." — Building an agent | Developer Documentation

📖 "which use different prompting strategies to execute tools." — Building an agent | Developer Documentation

from llama_index.llms.openai import OpenAI
from llama_index.core.agent.workflow import FunctionAgent

llm = OpenAI(model="gpt-4o-mini")
workflow = FunctionAgent(tools=[multiply, add], llm=llm, system_prompt="...")

📖 "you define a data structure in Pydantic and LlamaIndex works with Pydantic to coerce the output of the LLM into that structure." — Introduction to Structured Data Extraction | Developer Documentation

📖 "Pydantic is a widely-used data validation and conversion library. It relies heavily on Python type declarations." — Introduction to Structured Data Extraction | Developer Documentation

from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str = "Jane Doe"

📖 "This property is crucial: these JSON-formatted schemas are often passed to LLMs and the LLMs in turn use them as instructions on how to return data." — Introduction to Structured Data Extraction | Developer Documentation

📖 "To assist them and improve the accuracy of your returned data, it’s helpful to include natural-language descriptions of objects and fields and what they’re used for." — Introduction to Structured Data Extraction | Developer Documentation

{
  "properties": {
    "id": { "title": "Id", "type": "integer" },
    "name": { "default": "Jane Doe", "title": "Name", "type": "string" }
  },
  "required": ["id"],
  "title": "User",
  "type": "object"
}

📖 "You can create more complex structures by nesting these models:" — Introduction to Structured Data Extraction | Developer Documentation

📖 "Now Spam has a foo and a bars." — Introduction to Structured Data Extraction | Developer Documentation

from datetime import datetime


class LineItem(BaseModel):
    """A line item in an invoice."""

    item_name: str = Field(description="The name of this item")
    price: float = Field(description="The price of this item")


class Invoice(BaseModel):
    """A representation of information from an invoice."""

    invoice_id: str = Field(description="A unique identifier for this invoice, often a number")
    date: datetime = Field(description="The date this invoice was created")
    line_items: list[LineItem] = Field(description="A list of all the items in this invoice")

📖 "Each call to an LLM will cost some amount of money - for instance, OpenAI’s gpt-3.5-turbo costs $0.002 / 1k tokens." — use a mock llm globally

📖 "The following indices don’t require LLM calls at all during building (0 cost):" — use a mock llm globally

# 构建期零 LLM 成本的关键词表索引(用正则抽词)
from llama_index.core import SimpleKeywordTableIndex
index = SimpleKeywordTableIndex.from_documents(documents)

📖 "There will always be >= 1 LLM call during query time, in order to synthesize the final answer." — use a mock llm globally

📖 "SummaryIndex: by default requires {math}N LLM calls, where N is the number of nodes." — use a mock llm globally

📖 "Setting child_branch_factor=2 will be more expensive than the default child_branch_factor=1 (polynomial vs logarithmic), because we traverse 2 children instead of just 1 for each parent node." — use a mock llm globally

flowchart LR
    Q[查询] --> S[SummaryIndex: N 次]
    Q --> T[TreeIndex: log N 次]
    Q --> K[KeywordTableIndex: 1 次抽词]
    Q --> V[VectorStoreIndex: 1 次/查询]
    V -.调大 top_k/chunk_size.-> More[调用次数上升]

📖 "LlamaIndex offers token predictors to predict token usage of LLM and embedding calls." — use a mock llm globally

📖 "The max_tokens parameter is used as a “worst case” prediction, where each LLM response will contain exactly that number of tokens. If max_tokens is not specified, then it will simply predict back the prompt." — use a mock llm globally

from llama_index.core.llms import MockLLM
from llama_index.core import MockEmbedding
from llama_index.core import Settings

# use a mock llm globally
Settings.llm = MockLLM(max_tokens=256)
# use a mock embedding globally
Settings.embed_model = MockEmbedding(embed_dim=1536)

📖 "LlamaIndex also provides some tools / project templates to help you build a full-stack template." — Putting It All Together | Developer Documentation

📖 "We also have the llamaindex-cli rag CLI tool that combines some of the above concepts into an easy to use tool for chatting with files from your terminal!" — Putting It All Together | Developer Documentation

npx create-llama@latest

📖 "We have a guide on how to build a chatbot" — Putting It All Together | Developer Documentation

📖 "We have a complete guide to using property graphs for indexing and retrieval" — Putting It All Together | Developer Documentation

flowchart TD
    D{数据/交互形态} --> Text[非结构化文本] --> QA[Q&A / VectorStoreIndex]
    D --> SQL[结构化 SQL] --> ST[Structured Data 指南]
    D --> Graph[实体关系] --> PG[Property Graph 指南]
    D --> Chat[多轮对话] --> CB[Chatbot 指南]

flowchart LR
    A[数据源: PDF / API / DB] --> B[Document 通用容器]
    B -->|NodeParser 切分| C[Node chunk]
    C --> D[VectorStoreIndex]

📖 "Document and Node objects are core abstractions within LlamaIndex." — build index

📖 "is a generic container around any data source - for instance, a PDF, an API output, or retrieved data from a database." — build index

📖 "represents a “chunk” of a source Document, whether that is a text chunk, an image, or other." — build index

from llama_index.core import Document, VectorStoreIndex


text_list = [text1, text2, ...]
documents = [Document(text=t) for t in text_list]


# build index
index = VectorStoreIndex.from_documents(documents)

📖 "Nodes are a first-class citizen in LlamaIndex. You can choose to define Nodes and all its attributes directly." — build index

📖 "You may also choose to “parse” source Documents into Nodes through our NodeParser classes." — build index

📖 "By default every Node derived from a Document will inherit the same metadata from that Document" — build index

from llama_index.core.node_parser import SentenceSplitter


# load documents
...


# parse nodes
parser = SentenceSplitter()
nodes = parser.get_nodes_from_documents(documents)


# build index
index = VectorStoreIndex(nodes)

flowchart LR
    A[Sources] -->|loading| B[Documents]
    B -->|transformations 如切分| C[Nodes]
    C --> D[Index]

📖 "The key to data ingestion in LlamaIndex is loading and transformations. Once you have loaded Documents, you can process them via transformations and output Nodes." — Loading Data | Developer Documentation

📖 "The ingestion pipeline](/python/framework/module_guides/loading/ingestion_pipeline/index.md) which allows you to set up a repeatable, cache-optimized process for loading data." — Loading Data | Developer Documentation

📖 "By default, a Document stores text along with some other attributes." — build index

📖 "metadata - a dictionary of annotations that can be appended to the text." — build index

📖 "relationships - a dictionary containing relationships to other Documents/Nodes." — build index

📖 "You can choose to define Nodes and all its attributes directly." — build index

📖 "You may also choose to “parse” source Documents into Nodes through our NodeParser classes." — build index

parser = SentenceSplitter()
nodes = parser.get_nodes_from_documents(documents)

📖 "SimpleDirectoryReader](/python/framework/module_guides/loading/simpledirectoryreader/index.md), our built-in loader for loading all sorts of file types from a local directory" — Loading Data | Developer Documentation

📖 "LlamaParse](/python/framework/module_guides/loading/connector/llama_parse/index.md), LlamaIndex’s official tool for PDF parsing, available as a managed API." — Loading Data | Developer Documentation

📖 "LlamaHub](/python/framework/module_guides/loading/connector/index.md), our registry of hundreds of data loading libraries to ingest data from any source" — Loading Data | Developer Documentation

flowchart LR
  D[Documents] --> S[SentenceSplitter]
  S --> T[TitleExtractor]
  T --> E[OpenAIEmbedding]
  E --> N[Nodes]
  N --> V[(Vector Store)]
  S -.hash+cache.-> C[(IngestionCache)]
  T -.hash+cache.-> C
  E -.hash+cache.-> C

📖 "An IngestionPipeline uses a concept of Transformations that are applied to input data. These Transformations are applied to your input data, and the resulting nodes are either returned or inserted into a vector database (if given). Each node+transformation pair is cached, so that subsequent runs (if the cache is persisted) with the same node+transformation combination can use the cached result and save you time." — create the pipeline with transformations

📖 "If you are connecting your pipeline to a vector store, embeddings must be a stage of your pipeline or your later instantiation of the index will fail." — create the pipeline with transformations

from llama_index.core import Document
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.core.ingestion import IngestionPipeline, IngestionCache

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
        OpenAIEmbedding(),
    ]
)

nodes = pipeline.run(documents=[Document.example()])

📖 "Node parsers are a simple abstraction that take a list of documents, and chunk them into Node objects, such that each node is a specific chunk of the parent document. When a document is broken into nodes, all of it’s attributes are inherited to the children nodes (i.e. metadata, text and metadata templates, etc.)." — global

from llama_index.core import Document
from llama_index.core.node_parser import SentenceSplitter

node_parser = SentenceSplitter(chunk_size=1024, chunk_overlap=20)

nodes = node_parser.get_nodes_from_documents(
    [Document(text="long text")], show_progress=False
)

📖 "A data connector (aka Reader) ingest data from different data sources and data formats into a simple Document representation (text and simple metadata)." — Data Connectors (LlamaHub) | Developer Documentation

📖 "LlamaHub is an open-source repository containing data loaders that you can easily plug and play into any LlamaIndex application." — Data Connectors (LlamaHub) | Developer Documentation

from llama_index.core import download_loader

from llama_index.readers.google import GoogleDocsReader

loader = GoogleDocsReader()
documents = loader.load_data(document_ids=[...])

📖 "In an IngestionPipeline, each node + transformation combination is hashed and cached. This saves time on subsequent runs that use the same data." — create the pipeline with transformations

📖 "Here, no persist step is needed, since everything is cached as you go in the specified remote collection." — create the pipeline with transformations

# save
pipeline.persist("./pipeline_storage")

# load and restore state
new_pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        TitleExtractor(),
    ],
)
new_pipeline.load("./pipeline_storage")

# will run instantly due to the cache
nodes = pipeline.run(documents=[Document.example()])

📖 "Using the document.doc_id or node.ref_doc_id as a grounding point, the ingestion pipeline will actively look for duplicate documents." — create the pipeline with transformations

📖 "NOTE: If we do not attach a vector store, we can only check for and remove duplicate inputs." — create the pipeline with transformations

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[...], docstore=SimpleDocumentStore()
)

📖 "The run method of IngestionPipeline can be executed with parallel processes. It does so by making use of multiprocessing.Pool distributing batches of nodes to across processors." — create the pipeline with transformations

📖 "The IngestionPipeline also has support for async operation" — create the pipeline with transformations

from llama_index.core.ingestion import IngestionPipeline

pipeline = IngestionPipeline(
    transformations=[...],
)
pipeline.run(documents=[...], num_workers=4)

📖 "An Index is a data structure that allows us to quickly retrieve relevant context for a user query. For LlamaIndex, it’s the core foundation for retrieval-augmented generation (RAG) use-cases." — Indexing | Developer Documentation

📖 "Under the hood, Indexes store data in Node objects (which represent chunks of the original documents), and expose a Retriever interface that supports additional configuration and automation." — Indexing | Developer Documentation

flowchart LR
    D[Documents] --> N[Node 对象/chunks]
    N --> I[Index]
    I --> R[Retriever 接口]
    R --> QE[Query / Chat Engine]

📖 "Under the hood, LlamaIndex also supports swappable storage components that allows you to customize:" — construct vector store and customize storage context

📖 "The Document/Index stores rely on a common Key-Value store abstraction, which is also detailed below." — construct vector store and customize storage context

Document stores    -> ingested documents (Node objects)
Index stores       -> index metadata
Vector stores      -> embedding vectors
Property Graph     -> knowledge graphs (PropertyGraphIndex)
Chat Stores        -> chat messages
Docstore / Index store 都构建在共用的 Key-Value store 抽象之上

📖 "Many vector stores (except FAISS) will store both the data as well as the index (embeddings). This means that you will not need to use a separate document store or index store. This *also* means that you will not need to explicitly persist this data - this happens automatically." — construct vector store and customize storage context

📖 "Note that in general to use storage abstractions, you need to define a StorageContext object:" — construct vector store and customize storage context

## build a new index
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.deeplake import DeepLakeVectorStore


# construct vector store and customize storage context
vector_store = DeepLakeVectorStore(dataset_path="<dataset_path>")
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# Load documents and build index
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)




## reload an existing one
index = VectorStoreIndex.from_vector_store(vector_store=vector_store)

📖 "Note that in general to use storage abstractions, you need to define a StorageContext object:" — construct vector store and customize storage context

from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.storage.index_store import SimpleIndexStore
from llama_index.core.vector_stores import SimpleVectorStore
from llama_index.core import StorageContext


# create storage context using default stores
storage_context = StorageContext.from_defaults(
    docstore=SimpleDocumentStore(),
    vector_store=SimpleVectorStore(),
    index_store=SimpleIndexStore(),
)

📖 "LlamaIndex supports persisting data to any storage backend supported by fsspec. We have confirmed support for the following storage backends:" — construct vector store and customize storage context

- Local filesystem
- AWS S3
- Cloudflare R2
（凡 fsspec 支持的后端，理论上都可作为持久化目标）

📖 "Retrievers are responsible for fetching the most relevant context given a user query (or chat message)." — Retriever | Developer Documentation

📖 "It can be built on top of indexes, but can also be defined independently. It is used as a key building block in query engines (and Chat Engines) for retrieving relevant context." — Retriever | Developer Documentation

retriever = index.as_retriever()
nodes = retriever.retrieve("Who is Paul Graham?")

📖 "You can select the index-specific retriever class via retriever_mode." — Retriever | Developer Documentation

📖 "In the same way, you can pass kwargs to configure the selected retriever." — Retriever | Developer Documentation

retriever = summary_index.as_retriever(
    retriever_mode="llm",
    choice_batch_size=5,
)

📖 "You can use the low-level composition API if you need more granular control." — Retriever | Developer Documentation

from llama_index.core.retrievers import SummaryIndexLLMRetriever


retriever = SummaryIndexLLMRetriever(
    index=summary_index,
    choice_batch_size=5,
)

📖 "Querying is the most important part of your LLM application." — Querying | Developer Documentation

📖 "Otherwise check out how to use our query modules as standalone components 👇." — Querying | Developer Documentation

flowchart LR
    Q[User Query] --> R[Retriever]
    R --> NP[Node Postprocessors]
    NP --> RS[Response Synthesizers]
    RS --> SO[Structured Outputs]
    RT[Routers] -.选择路径.-> R
    QE[Query / Chat Engines / Agents] -.编排.-> R

📖 "You can create workflows for querying with ease, using our event-driven Workflow interface." — Querying | Developer Documentation

flowchart TD
    Start[Query Event] --> Retrieve[Retrieve Nodes]
    Retrieve --> Post[Postprocess]
    Post --> Synth[Synthesize Response]
    Synth --> End[Result Event]

📖 "refine: *create and refine* an answer by sequentially going through each retrieved text chunk. This makes a separate LLM call per Node/retrieved chunk." — NOTE: we add an extra tone_name variable here

📖 "compact (default): similar to refine but *compact* (concatenate) the chunks beforehand, resulting in less LLM calls." — NOTE: we add an extra tone_name variable here

📖 "simple_summarize: Truncates all text chunks to fit into a single LLM prompt. Good for quick summarization purposes, but may lose detail due to truncation." — NOTE: we add an extra tone_name variable here

from llama_index.core import get_response_synthesizer

# 摘要类问题优先用 tree_summarize；追求低成本可换成 "compact"
response_synthesizer = get_response_synthesizer(response_mode="tree_summarize")
query_engine = index.as_query_engine(response_synthesizer=response_synthesizer)
response = query_engine.query("query_text")

📖 "Node postprocessors are a set of modules that take a set of nodes, and apply some kind of transformation or filtering before returning them." — similarity postprocessor: filter nodes below 0.75 similarity score

📖 "In LlamaIndex, node postprocessors are most commonly applied within a query engine, after the node retrieval step and before the response synthesis step." — similarity postprocessor: filter nodes below 0.75 similarity score

flowchart LR
  Q[User Query] --> R[Retriever 召回 nodes]
  R --> P[Node Postprocessor 过滤/重排]
  P --> S[Response Synthesis 合成答案]
  S --> A[Response]

from llama_index.core.postprocessor import SimilarityPostprocessor

nodes = index.as_retriever().retrieve("test query str")
# filter nodes below 0.75 similarity score
processor = SimilarityPostprocessor(similarity_cutoff=0.75)
filtered_nodes = processor.postprocess_nodes(nodes)

📖 "A Response Synthesizer is what generates a response from an LLM, using a user query and a given set of text chunks. The output of a response synthesizer is a Response object." — NOTE: we add an extra tone_name variable here

📖 "When used in a query engine, the response synthesizer is used after nodes are retrieved from a retriever, and after any node-postprocessors are ran." — NOTE: we add an extra tone_name variable here

from llama_index.core.data_structs import Node
from llama_index.core.response_synthesizers import ResponseMode
from llama_index.core import get_response_synthesizer

response_synthesizer = get_response_synthesizer(
    response_mode=ResponseMode.COMPACT
)

response = response_synthesizer.synthesize(
    "query text", nodes=[Node(text="text"), ...]
)

📖 "The base class is BaseNodePostprocessor, and the API interface is very simple:" — similarity postprocessor: filter nodes below 0.75 similarity score

📖 "the postprocessors take NodeWithScore objects as inputs, which is just a wrapper class with a Node and a score value." — similarity postprocessor: filter nodes below 0.75 similarity score

from llama_index.core import QueryBundle
from llama_index.core.postprocessor.types import BaseNodePostprocessor
from llama_index.core.schema import NodeWithScore


class DummyNodePostprocessor(BaseNodePostprocessor):
    def _postprocess_nodes(
        self, nodes: List[NodeWithScore], query_bundle: Optional[QueryBundle]
    ) -> List[NodeWithScore]:
        # subtracts 1 from the score
        for n in nodes:
            n.score -= 1

        return nodes

📖 "With structured_answer_filtering set to True, our refine module is able to filter out any input nodes that are not relevant to the question being asked." — NOTE: we add an extra tone_name variable here

📖 "Other LLM providers or models that don’t have native function calling support may be less reliable in producing the structured response this feature relies on." — NOTE: we add an extra tone_name variable here

from llama_index.core import get_response_synthesizer

response_synthesizer = get_response_synthesizer(structured_answer_filtering=True)

📖 "You may want to customize the prompts used in our response synthesizer, and also add additional variables during query-time." — NOTE: we add an extra tone_name variable here

📖 "You can specify these additional variables in the **kwargs for get_response." — NOTE: we add an extra tone_name variable here

from llama_index.core import PromptTemplate
from llama_index.core.response_synthesizers import TreeSummarize


# NOTE: we add an extra tone_name variable here
qa_prompt_tmpl = (
    "Context information is below.\n"
    "---------------------\n"
    "{context_str}\n"
    "---------------------\n"
    "Given the context information and not prior knowledge, "
    "answer the query.\n"
    "Please also write the answer in the tone of {tone_name}.\n"
    "Query: {query_str}\n"
    "Answer: "
)
qa_prompt = PromptTemplate(qa_prompt_tmpl)

# initialize response synthesizer
summarizer = TreeSummarize(verbose=True, summary_template=qa_prompt)

# get response
response = summarizer.get_response(
    "who is Paul Graham?", [text], tone_name="a Shakespeare play"
)

📖 "Routers are modules that take in a user query and a set of “choices” (defined by metadata), and returns one or more selected choices." — pydantic selectors feed in pydantic objects to a function calling API

📖 "They are simple but powerful modules that use LLMs for decision making capabilities." — pydantic selectors feed in pydantic objects to a function calling API

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import PydanticSingleSelector
from llama_index.core.tools import QueryEngineTool

list_tool = QueryEngineTool.from_defaults(
    query_engine=list_query_engine,
    description="Useful for summarization questions related to the data source",
)
vector_tool = QueryEngineTool.from_defaults(
    query_engine=vector_query_engine,
    description="Useful for retrieving specific context related to the data source",
)

query_engine = RouterQueryEngine(
    selector=PydanticSingleSelector.from_defaults(),
    query_engine_tools=[list_tool, vector_tool],
)
query_engine.query("<query>")

📖 "LLM selectors put the choices as a text dump into a prompt and use LLM text completion endpoint to make decisions" — pydantic selectors feed in pydantic objects to a function calling API

📖 "Pydantic selectors pass choices as Pydantic schemas into a function calling endpoint, and return Pydantic objects" — pydantic selectors feed in pydantic objects to a function calling API

from llama_index.core.selectors import LLMSingleSelector, LLMMultiSelector
from llama_index.core.selectors import (
    PydanticMultiSelector,
    PydanticSingleSelector,
)

# pydantic selectors feed in pydantic objects to a function calling API
selector = PydanticSingleSelector.from_defaults()
selector = PydanticMultiSelector.from_defaults()

# LLM selectors use text completion endpoints
selector = LLMSingleSelector.from_defaults()
selector = LLMMultiSelector.from_defaults()

📖 "You can use the selectors as standalone modules. Define choices as either a list of ToolMetadata or as a list of strings." — pydantic selectors feed in pydantic objects to a function calling API

from llama_index.core.tools import ToolMetadata
from llama_index.core.selectors import LLMSingleSelector

# choices as a list of tool metadata
choices = [
    ToolMetadata(description="description for choice 1", name="choice_1"),
    ToolMetadata(description="description for choice 2", name="choice_2"),
]

# choices as a list of strings
choices = [
    "choice 1 - description for choice 1",
    "choice 2: description for choice 2",
]

selector = LLMSingleSelector.from_defaults()
selector_result = selector.select(
    choices, query="What's revenue growth for IBM in 2007?"
)
print(selector_result.selections)

📖 "The ability of LLMs to produce structured outputs are important for downstream applications that rely on reliably parsing output values." — Structured Outputs | Developer Documentation

📖 "Output Parsers: These are modules that operate before and after an LLM text completion endpoint. They are not used with LLM function calling endpoints (since those contain structured outputs out of the box)." — Structured Outputs | Developer Documentation

📖 "With generic completion APIs, the inputs and outputs are handled by text prompts. The output parser plays a role before and after the LLM call in ensuring structured outputs." — Structured Outputs | Developer Documentation

📖 "With function calling APIs, the output is inherently in a structured format, and the input can take in the signature of the desired object." — Structured Outputs | Developer Documentation

flowchart TD
    Q[输入 prompt] --> B{哪种 API?}
    B -->|completion| P1[output parser 追加 format instructions]
    P1 --> L1[LLM text completion]
    L1 --> P2[output parser 解析输出]
    P2 --> O1[结构化结果]
    B -->|function calling| L2[LLM function calling]
    L2 --> C[cast 成 Pydantic 对象]
    C --> O2[结构化结果]

📖 "Prompting is the fundamental input that gives LLMs their expressive power. LlamaIndex uses prompts to build the index, do insertion, perform traversal during querying, and to synthesize the final answer." — you can create text prompt (for completion API)

📖 "When building agentic workflows, building and managing prompts is a key part of the development process. LlamaIndex provides a flexible and powerful way to manage prompts, and to use them in a variety of ways." — you can create text prompt (for completion API)

from llama_index.core.prompts import RichPromptTemplate

template_str = """We have provided context information below.
---------------------
{{ context_str }}
---------------------
Given this information, please answer the question: {{ query_str }}
"""
qa_template = RichPromptTemplate(template_str)

📖 "RichPromptTemplate - latest-style for building jinja-style prompts with variables and logic" — you can create text prompt (for completion API)

📖 "PromptTemplate - older-style simple templating for building prompts with a single f-string" — you can create text prompt (for completion API)

📖 "ChatPromptTemplate - older-style simple templating for building chat prompts with messages and f-strings" — you can create text prompt (for completion API)

flowchart TD
    A[需要构建 prompt] --> B{是否需要变量+逻辑?}
    B -->|需要 jinja 逻辑| C[RichPromptTemplate]
    B -->|简单文本插值| D{面向哪种 API?}
    D -->|completion 单串文本| E[PromptTemplate f-string]
    D -->|chat 多角色消息| F[ChatPromptTemplate messages]

📖 "Using prompts is simple. Below is an example of how to use the RichPromptTemplate to build a jinja-style prompt template:" — you can create text prompt (for completion API)

# you can create text prompt (for completion API)
prompt = qa_template.format(context_str=..., query_str=...)


# or easily convert to message prompts (for chat API)
messages = qa_template.format_messages(context_str=..., query_str=...)

📖 "Users may also provide their own prompt templates to further customize the behavior of the framework. The best method for customizing is copying the default prompt from the link above, and using that as the base for any modifications." — you can create text prompt (for completion API)

📖 "In addition, there are some prompts written and used specifically for chat models like gpt-3.5-turbo here." — you can create text prompt (for completion API)

# 1) 打开默认模板源码,复制目标 prompt
# https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/prompts/default_prompts.py
# 2) 以复制内容为 base,只改需要变动的部分
# 3) 用 RichPromptTemplate / PromptTemplate 重新实例化后传入框架

📖 "There are a few primary ways you interact with models in LlamaIndex:" — Models | Developer Documentation

flowchart LR
    D[原始文档] -->|Embeddings 向量化| I[索引/向量库]
    Q[用户问题] -->|Embeddings 向量化| R[检索]
    I --> R
    R -->|上下文| L[LLMs 生成答案]
    M[MultiModal 实验性] -.图文输入.-> L

📖 "Query engine is a generic interface that allows you to ask question over your data." — Query Engine | Developer Documentation

📖 "A query engine takes in a natural language query, and returns a rich response. It is most often (but not always) built on one or many indexes via retrievers. You can compose multiple query engines to achieve more advanced capability." — Query Engine | Developer Documentation

query_engine = index.as_query_engine(streaming=True)
streaming_response = query_engine.query("Who is Paul Graham.")
streaming_response.print_response_stream()

📖 "Chat engine is a high-level interface for having a conversation with your data (multiple back-and-forth instead of a single question & answer). Think ChatGPT, but augmented with your knowledge base." — Chat Engine | Developer Documentation

📖 "Conceptually, it is a stateful analogy of a Query Engine. By keeping track of the conversation history, it can answer questions with past context in mind." — Chat Engine | Developer Documentation

chat_engine = index.as_chat_engine()
streaming_response = chat_engine.stream_chat("Tell me a joke.")
for token in streaming_response.response_gen:
    print(token, end="")

📖 "In LlamaIndex, we define an “agent” as a specific system that uses an LLM, memory, and tools, to handle inputs from outside users. Contrast this with the term “agentic”, which generally refers to a superclass of agents, which is any system with LLM decision making in the process." — Define a simple calculator tool

📖 "The FunctionAgent is a type of agent that uses an LLM provider’s function/tool calling capabilities to execute tools. Other types of agents, such as ReActAgent and CodeActAgent, use different prompting strategies to execute tools." — Define a simple calculator tool

flowchart TD
    A[获取最新消息 + chat history] --> B[将 tool schemas 和 history 发给 API]
    B --> C{LLM 直接回答
还是返回 tool calls?}
    C -->|直接回答| D[返回最终 response]
    C -->|tool calls| E[执行每个 tool call]
    E --> F[结果追加进 chat history]
    F --> A

📖 "Memory is a core-component when building agents. By default, all LlamaIndex agents are using a ChatMemoryBuffer for memory." — Define a simple calculator tool

from llama_index.core.memory import ChatMemoryBuffer


memory = ChatMemoryBuffer.from_defaults(token_limit=40000)


response = await agent.run(..., memory=memory)

📖 "Tools can be defined simply as python functions, or further customized using classes like FunctionTool and QueryEngineTool. LlamaIndex also provides sets of pre-defined tools for common APIs using something called Tool Specs." — Define a simple calculator tool

def multiply(a: float, b: float) -> float:
    """Useful for multiplying two numbers."""
    return a * b


agent = FunctionAgent(
    tools=[multiply],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="You are a helpful assistant that can multiply two numbers.",
)

📖 "You can combine agents into a multi-agent system, where each agent is able to hand off control to another agent to coordinate while completing tasks." — Define a simple calculator tool

from llama_index.core.agent.workflow import AgentWorkflow


multi_agent = AgentWorkflow(agents=[FunctionAgent(...), FunctionAgent(...)])


resp = await agent.run("query")

📖 "Does the response match the retrieved context? Does it also match the query? Does it match the reference answer or guidelines?" — Evaluating | Developer Documentation

📖 "Are the retrieved sources relevant to the query?" — Evaluating | Developer Documentation

📖 "given a dataset of questions and ground-truth rankings, we can evaluate retrievers using ranking metrics like mean-reciprocal rank (MRR), hit-rate, precision, and more." — Evaluating | Developer Documentation

flowchart LR
    A[unstructured text corpus] --> B[Dataset generation
合成 question,context 对]
    B --> C[Retrieval Evaluation
用 MRR/hit-rate 打分]
    C --> D[定位: retriever 问题?]

📖 "This uses a “gold” LLM (e.g. GPT-4) to decide whether the predicted answer is correct in a variety of ways." — Evaluating | Developer Documentation

📖 "Evaluation can be done with some combination of the query, context, response, and combine these with LLM calls." — Evaluating | Developer Documentation

📖 "Evaluates if the answer is faithful to the retrieved contexts (in other words, whether if there’s hallucination)." — Evaluating | Developer Documentation

需要 labels: Correctness, Semantic Similarity
不需要 labels: Faithfulness, Context Relevancy, Answer Relevancy, Guideline Adherence

📖 "LlamaIndex provides one-click observability 🔭 to allow you to build principled LLM applications in a production setting." — general usage

📖 "Note that all kwargs to set_global_handler are passed to the underlying callback handler." — general usage

📖 "A lot of the tooling and integrations mentioned in this page use our legacy CallbackManager or don’t use set_global_handler." — general usage

import llama_index.core

llama_index.core.set_global_handler("simple")

📖 "LlamaIndex provides callbacks to help debug, track, and trace the inner workings of the library. Using the callback manager, as many callbacks as needed can be added." — Callbacks | Developer Documentation

📖 "In addition to logging data related to events, you can also track the duration and number of occurrences of each event." — Callbacks | Developer Documentation

📖 "For example, the LlamaDebugHandler will, by default, print the trace of events after most operations." — Callbacks | Developer Documentation

QUERY(start) -> RETRIEVE(召回 nodes) -> SYNTHESIZE -> LLM(模板+响应) -> QUERY(end)

📖 "Model Context Protocol (MCP) is an open-source standard protocol that allows Large Language Models (LLMs) to interact with external tools and data sources through structured API calls." — Model Context Protocol (MCP) | Developer Documentation

📖 "Think of MCP as a “USB-C port” for AI applications - it provides a standardized way for various tools, platforms, and data sources to connect to AI models." — Model Context Protocol (MCP) | Developer Documentation

📖 "Lightweight services that expose capabilities (tools, resources, prompts) via the standardized protocol" — Model Context Protocol (MCP) | Developer Documentation

flowchart LR
    H[MCP Host
Claude Desktop / IDE] --> C[MCP Client
1:1 连接]
    C -->|structured API calls| S[MCP Server
暴露 tools/resources/prompts]
    S --> R[外部工具 / 数据库 / 模板]

📖 "Get data from external resources that are served via existing MCP servers." — Model Context Protocol (MCP) | Developer Documentation

📖 "You can convert your own custom LlamaIndex workflows to MCP servers." — Model Context Protocol (MCP) | Developer Documentation

1. 消费已有 MCP server 的 tools  -> LlamaIndex 作为 client
2. 把自己的 workflow 变成 MCP server -> LlamaIndex 作为 server
3. 跑 LlamaCloud MCP server (Py/TS)  -> 桥接 LlamaExtract/LlamaParse 到任意 MCP 应用

📖 "A workflow is an event-driven, step-based way to control the execution flow of an application." — pip install llama-index-llms-openai if you don't already have it

📖 "A step receives an event, does some work, and returns another event. That returned event triggers the next step whose type annotation accepts it." — pip install llama-index-llms-openai if you don't already have it

flowchart LR
    Start[StartEvent] --> G[generate_joke]
    G --> J[JokeEvent]
    J --> C[critique_joke]
    C --> Stop[StopEvent]

📖 "Logic like loops and branches needed to be encoded into the edges of graphs, which made them hard to read and understand." — pip install llama-index-llms-openai if you don't already have it

📖 "Branches are ordinary if statements that return different event types. Loops are steps that return an event handled by an earlier step. Concurrent work is a step that returns list[Event], paired with another step that accepts list[Event]." — pip install llama-index-llms-openai if you don't already have it

@step
async def reflect(self, ev: DraftEvent) -> DraftEvent | StopEvent:
    # 校验不通过就返回 DraftEvent，被更早的生成 step 再次处理，形成循环
    ...

📖 "When the workflow encounters a returned StopEvent, it immediately stops the workflow and returns whatever we passed in the result parameter." — pip install llama-index-llms-openai if you don't already have it

📖 "The .run() method starts the workflow and returns a WorkflowHandler. The handler is awaitable, so await w.run(...) waits for the final result." — pip install llama-index-llms-openai if you don't already have it

handler = w.run(topic="pirates")
async for ev in handler.stream_events():
    ...
result = await handler

📖 "Before a workflow runs, Workflows validates the event graph described by your step signatures. It checks that start and stop events are present, produced events have consumers, consumed events have producers, and the graph does not contain accidental dead ends." — pip install llama-index-llms-openai if you don't already have it

@step(skip_graph_checks=["reachability"])
async def receive_webhook(self, ev: WebhookEvent) -> StopEvent:
    ...

📖 "Most workflow code should use typed step inputs and typed returns. That gives validation, diagrams, and readable code" — pip install llama-index-llms-openai if you don't already have it

📖 "The type-first APIs are easier to validate and visualize. The context APIs are more flexible, but they make you own more of the bookkeeping." — pip install llama-index-llms-openai if you don't already have it

Return `A | B`            -> 分支到不同 event 类型
Return `list[A]`          -> 有限一批工作,先全部产出再交给下游
Accept `list[A]`          -> 需要整批结果后再继续
ctx.send_event(...)       -> 增量/未知数量/外部发送 event
ctx.collect_events(...)   -> 手动等待一组已知 event
ctx.store                 -> step 间每次运行的共享状态
Resource(...)             -> 不该进序列化状态的依赖

📖 "LlamaCloud is a managed platform for data parsing, ingestion, retrieval, structured data extraction, and more. LlamaCloud enables you to get production-quality data for your production LLM application." — LlamaCloud | Developer Documentation

📖 "LlamaParse is the enterprise platform for turning documents into production AI pipelines. One API key, one SDK, and six composable products: Parse (agentic OCR), Extract (structured data), Classify, Split, Sheets, and Index." — Upload and parse a document

pip install llama-cloud>=2.1
export LLAMA_CLOUD_API_KEY=llx-...

from llama_cloud import LlamaCloud

client = LlamaCloud()  # Uses LLAMA_CLOUD_API_KEY env var

# Upload and parse a document
file = client.files.create(file="document.pdf", purpose="parse")
result = client.parsing.parse(
    file_id=file.id,
    tier="agentic",
    version="latest",
    expand=["markdown"],
)

📖 "Most parsers are bounding-box detectors bolted onto an OCR engine. Parse is different. It uses generative models for end-to-end document understanding, so layout, structure, tables, charts, and text all come out of the same reasoning pass—instead of getting stitched back together from dozens of fragile heuristics. That’s why it handles complex real-world documents where traditional parsers break down." — Overview of Parse | Developer Documentation

📖 "For a single short document, passing the raw PDF to a multimodal model often works fine. At scale it breaks down: tables lose structure, scanned pages produce noisy OCR, and token cost scales linearly with raw page content regardless of whether it carried useful signal. Parse does the structural work once, upstream, so downstream models only see clean, layout-aware input." — Overview of Parse | Developer Documentation

from llama_cloud import LlamaCloud

client = LlamaCloud()  # reads LLAMA_CLOUD_API_KEY
file = client.files.create(file="doc.pdf", purpose="parse")
result = client.parsing.parse(file_id=file.id, tier="agentic", version="latest")
print(result.markdown.pages[0].markdown)

flowchart LR
    A[原始文档 PDF/扫描/图表] --> B{解析范式}
    B -->|传统: OCR + bbox + 启发式缝合| C[版面/表格易崩]
    B -->|Parse: 生成式端到端一次推理| D[干净 layout-aware markdown/text/JSON]
    D --> E[下游 LLM 只看干净输入]

📖 "Fast — lowest-latency, lowest-cost tier for plain-text documents at high volume; returns text and spatial text only (no markdown)." — Overview of Parse | Developer Documentation

📖 "Cost Optimizer — pay premium prices only on the pages that need it. Parse routes each page in your document to the right tier automatically: simple pages go to the cheaper cost_effective tier, complex pages (tables, charts, multi-column layouts, diagrams) stay on your premium tier, and both run in parallel so there’s no latency penalty." — Overview of Parse | Developer Documentation

📖 "LlamaExtract provides a simple API for extracting structured data from unstructured documents like PDFs, text files, and images." — Overview of Extract | Developer Documentation

📖 "LlamaExtract guarantees that your data complies with the provided schema or provides helpful error messages when it doesn’t." — Overview of Extract | Developer Documentation

from pydantic import BaseModel, Field
from llama_cloud import LlamaCloud

# Define your schema
class Resume(BaseModel):
    name: str = Field(description="Full name of candidate")
    email: str = Field(description="Email address")
    skills: list[str] = Field(description="Technical skills")

client = LlamaCloud()

# Upload and extract
file = client.files.create(file="resume.pdf", purpose="extract")
job = client.extract.create(
    document_input_value=file.id,
    config={
        "extract_options": {
            "data_schema": Resume.model_json_schema(),
            "tier": "agentic",
        },
    },
)
print(job.extract_result)

📖 "For production workflows, pin version to the date you create or update the configuration in YYYY-MM-DD format, for example 2026-03-31. A date pin resolves to the most recent available extract version for the selected tier at or before that date, so later releases do not silently change existing behavior. Use latest when you want jobs to automatically pick up the newest extract version." — Overview of Extract | Developer Documentation

📖 "When using the SDK or REST API directly, V2 decouples parse and extract tiers." — Overview of Extract | Developer Documentation

V2 extract tier   V2 parse tier    V1 equivalent (extraction_mode)
cost_effective    fast             FAST
agentic           agentic          MULTIMODAL
agentic           agentic_plus     PREMIUM

📖 "nodes = run_transformations(" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/base.py

📖 "The constructor now takes in a list of Node objects. " — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/base.py

flowchart TD
    A[Document 列表] -->|set_document_hash| B[记录文档哈希]
    A -->|run_transformations| C[Sequence of BaseNode]
    C --> D["cls(nodes=nodes, ...)"]
    D -->|__init__| E{"index_struct is None?"}
    E -->|是| F["build_index_from_nodes(nodes + objects)"]
    F --> G[IndexDict]
    G --> H["index_store.add_index_struct"]

📖 "for nodes_batch in iter_batch(nodes, self._insert_batch_size):" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "Embeddings are called in batches." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "new_ids = self._vector_store.add(nodes_batch, **insert_kwargs)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

# 构造时可自定义批大小,大批量导入时调大可减少往返
index = VectorStoreIndex(
    nodes=nodes,
    insert_batch_size=2048,  # 默认值
    show_progress=True,
)
# 内部等价于对每个 batch:
#   nodes_batch = self._get_node_with_embedding(nodes_batch, show_progress)
#   new_ids = self._vector_store.add(nodes_batch, **insert_kwargs)

📖 "if not self._vector_store.stores_text or self._store_nodes_override:" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "# NOTE: remove embedding from node to avoid duplication" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "Vector store integrations that store text in the vector store are " — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

# 向量库存文本时,仍想保留本地 Node(以便 ref_doc_info / 删除)
index = VectorStoreIndex(
    nodes=nodes,
    store_nodes_override=True,   # 强制写 index_struct + docstore
)
# 否则 index.ref_doc_info 在 stores_text=True 时会 raise NotImplementedError

📖 "if self._use_async:" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "run_async_tasks(tasks)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "Whether to use asynchronous calls. Defaults to False." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

# 大规模导入时开启异步,让 embedding / 写入并发
index = VectorStoreIndex(
    nodes=nodes,
    use_async=True,   # 走 _async_add_nodes_to_index + run_async_tasks
)

📖 "Cannot initialize from a vector store that does not store text." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "storage_context = StorageContext.from_defaults(vector_store=vector_store)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

# 向量库里已有 embedding + 文本,直接挂载检索
index = VectorStoreIndex.from_vector_store(
    vector_store=my_vector_store,   # 必须 stores_text=True
    embed_model=my_embed_model,
)
retriever = index.as_retriever()

📖 "print("Some nodes are missing content, skipping them...")" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/base.py

📖 "This is equivalent to deleting the document and then inserting it again." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/base.py

📖 "existing_doc_hash = self._docstore.get_document_hash(document.id_)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/base.py

# refresh 只对哈希变化的文档重建,返回每个文档是否被刷新
refreshed = index.refresh_ref_docs(documents)
# refreshed[i] == True 表示第 i 个文档是新插入或因内容变化被 update

📖 " if isinstance(str_or_query_bundle, str):

📖 " nodes = self._retrieve(query_bundle)

# 两种调用最终在 retrieve 内部都归一化为 QueryBundle
retriever.retrieve("什么是向量检索?")
retriever.retrieve(QueryBundle(query_str="什么是向量检索?"))

flowchart TD
    A["retrieve(str_or_query_bundle)"] --> B["_check_callback_manager"]
    B --> C["RetrievalStartEvent"]
    C --> D{"是 str?"}
    D -->|是| E["QueryBundle(str)"]
    D -->|否| F["直接用 QueryBundle"]
    E --> G["_retrieve(query_bundle)"]
    F --> G
    G --> H["_handle_recursive_retrieval"]
    H --> I["RetrievalEndEvent + 返回 nodes"]

📖 " obj = node.obj or self.object_map.get(node.index_id, None)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/base_retriever.py

📖 " # remove any duplicates based on node_id

if objects is not None:
    object_map = {obj.index_id: obj.obj for obj in objects}

📖 " elif isinstance(obj, BaseQueryEngine):

📖 " raise ValueError(f"Object {obj} is not retrievable.")" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/base_retriever.py

# _retrieve_from_object 内部
elif isinstance(obj, BaseRetriever):
    return obj.retrieve(query_bundle)   # 递归进入子检索器的完整 retrieve 流程

📖 " return (

📖 " if query_bundle.embedding is None and len(query_bundle.embedding_strs) > 0:" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/retrievers/retriever.py

retriever = index.as_retriever(
    vector_store_query_mode="sparse",   # _needs_embedding() 返回 False
    similarity_top_k=5,
)

📖 " If the vector store does not store text, we need to fetch every node from the docstore." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/indices/vector_store/retrievers/retriever.py

📖 " fetched_nodes: List[BaseNode] = self._docstore.get_nodes(

# _get_nodes_with_embeddings 内
nodes_to_fetch = self._determine_nodes_to_fetch(query_result)
if nodes_to_fetch:
    fetched_nodes = self._docstore.get_nodes(node_ids=nodes_to_fetch, raise_error=False)
    query_result.nodes = self._insert_fetched_nodes_into_query_result(query_result, fetched_nodes)

📖 " # TODO: make this abstract

📖 " return self._retrieve(query_bundle)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/base_retriever.py

async def _aget_nodes_with_embeddings(self, query_bundle_with_embeddings):
    query = self._build_vector_store_query(query_bundle_with_embeddings)
    query_result = await self._vector_store.aquery(query, **self._kwargs)
    ...

flowchart LR
  Q[query_bundle] --> E[callback QUERY event]
  E --> R["self.retrieve()
检索+后处理"]
  R --> S["response_synthesizer.synthesize()
LLM 合成答案"]
  S --> O[query_event.on_end]
  O --> Resp[RESPONSE_TYPE]

📖 " with self.callback_manager.event(

nodes = self.retrieve(query_bundle)          # 检索并跑完所有 node_postprocessors
response = self._response_synthesizer.synthesize(query=query_bundle, nodes=nodes)
# 整个过程被 CBEventType.QUERY 事件包裹,记录 query_str 与 response

📖 " def retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:

📖 " for node_postprocessor in self._node_postprocessors:

engine = RetrieverQueryEngine(
    retriever=retriever,
    node_postprocessors=[reranker, SimilarityPostprocessor(similarity_cutoff=0.7)],
)
# retrieve 内部: nodes = retriever.retrieve(...) -> reranker -> similarity_cutoff

📖 " response_mode: ResponseMode = ResponseMode.COMPACT," — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/query_engine/retriever_query_engine.py

📖 " response_synthesizer = response_synthesizer or get_response_synthesizer(

engine = RetrieverQueryEngine.from_args(
    retriever=retriever,
    response_mode=ResponseMode.REFINE,   # 覆盖默认的 COMPACT
    streaming=True,
    verbose=True,
)

📖 " nodes = await self.aretrieve(query_bundle)

📖 " async def aretrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:

response = await engine.aquery("What is X?")   # 触发 _aquery -> aretrieve -> asynthesize

📖 " def with_retriever(self, retriever: BaseRetriever) -> "RetrieverQueryEngine":

📖 " def synthesize(

new_engine = engine.with_retriever(hybrid_retriever)   # synthesizer / postprocessors 全部沿用
# 或者拿外部节点直接合成,跳过检索:
resp = engine.synthesize(query_bundle, nodes=my_prefetched_nodes)

flowchart LR
    A[chunk1] -->|text_qa_template| B[answer_v1]
    B -->|refine_template + existing_answer| C[answer_v2]
    C -->|refine_template + existing_answer| D[answer_v3]
    D --> E[最终答案]

📖 """Refine a response to a query across text chunks.""" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

📖 query_str=query_str, existing_answer=response — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

📖 # If chunk is too big to be packed into a single chunk, push new chunks into the front of the deque — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

from llama_index.core import get_response_synthesizer
synth = get_response_synthesizer(response_mode="refine")
# chunk0 -> QA 得到初稿; chunk1、chunk2 分别带着 existing_answer 走 refine
answer = synth.get_response(query_str="LlamaIndex 是什么?", text_chunks=[c0, c1, c2])

📖 """Refine responses across compact text chunks.""" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/compact_and_refine.py

📖 # use prompt helper to fix compact text_chunks under the prompt limitation — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/compact_and_refine.py

📖 response_mode: ResponseMode = ResponseMode.COMPACT, — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/factory.py

from llama_index.core import get_response_synthesizer
# 不传 response_mode 时,factory 默认走 ResponseMode.COMPACT => CompactAndRefine
synth = get_response_synthesizer()  # 等价于 response_mode="compact"
answer = synth.get_response(query_str="...", text_chunks=[c0, c1, c2])

flowchart BT
    c1[chunk1] --> s1[摘要1]
    c2[chunk2] --> s2[摘要2]
    c3[chunk3] --> s3[摘要3]
    s1 --> R[递归汇总]
    s2 --> R
    s3 --> R
    R --> F[最终响应]

📖 in a bottom-up fashion (i.e. building a tree from leaves to root). — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

📖 # give final response if there is only one chunk — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

📖 # recursively summarize the summaries — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

from llama_index.core import get_response_synthesizer
synth = get_response_synthesizer(response_mode="tree_summarize", use_async=True)
summary = synth.get_response(query_str="总结这些文档的要点", text_chunks=chunks)

📖 description="True if there was enough context given to provide an answer " — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

📖 query_satisfied=True. In effect, doesn't do any answer filtering. — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

📖 "Program factory not supported without structured answer filtering." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/refine.py

from llama_index.core import get_response_synthesizer
synth = get_response_synthesizer(
    response_mode="refine",
    structured_answer_filtering=True,  # 无关块会被 query_satisfied=False 过滤
)

📖 response_mode: ResponseMode = ResponseMode.COMPACT, — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/factory.py

📖 raise ValueError(f"Unknown mode: {response_mode}") — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/factory.py

from llama_index.core import get_response_synthesizer
r = get_response_synthesizer(response_mode="refine")          # -> Refine
c = get_response_synthesizer()                                # -> CompactAndRefine(默认)
t = get_response_synthesizer(response_mode="tree_summarize")  # -> TreeSummarize

📖 # summarize each chunk — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

📖 summary_responses = run_async_tasks(tasks) — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

📖 summaries = await asyncio.gather(*str_tasks) — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

from llama_index.core import get_response_synthesizer
synth = get_response_synthesizer(response_mode="tree_summarize", use_async=True)
# 内部对每层多个 chunk 的摘要走 run_async_tasks 并发,而非逐块串行
resp = synth.get_response(query_str="概述全部文档", text_chunks=many_chunks)

flowchart TD
    A[text] --> B{token_size <= chunk_size?}
    B -- yes --> C[return 单个 _Split is_sentence=True]
    B -- no --> D[_split_fns: 段落分隔符 / 句子 tokenizer]
    D -- 切出 >1 片 --> E[is_sentence=True]
    D -- 切不动 --> F[_sub_sentence_split_fns: 次级正则 / word / char]
    F --> G[is_sentence=False]
    E --> H{片段仍超 chunk_size?}
    G --> H
    H -- yes --> D
    H -- no --> I[加入 text_splits]

📖 "In general, this class tries to keep sentences and paragraphs together. Therefore

📖 "The order of splitting is:

def _get_splits_by_fns(self, text: str) -> Tuple[List[str], bool]:
    for split_fn in self._split_fns:
        splits = split_fn(text)
        if len(splits) > 1:
            return splits, True

    for split_fn in self._sub_sentence_split_fns:
        splits = split_fn(text)
        if len(splits) > 1:
            break

    return splits, False

📖 "if cur_split.token_size > chunk_size:

📖 " # add overlap to the next chunk using the last one first

if (
    cur_split.is_sentence
    or cur_chunk_len + cur_split.token_size <= chunk_size
    or new_chunk  # new chunk, always add at least one split
):
    # add split to chunk
    cur_chunk_len += cur_split.token_size
    cur_chunk.append((cur_split.text, cur_split.token_size))
    split_idx += 1
    new_chunk = False
else:
    # close out chunk
    close_chunk()

📖 " metadata_len = len(self._tokenizer(metadata_str))

📖 " # use the longest metadata str for splitting

elif effective_chunk_size < 50:
    print(
        f"Metadata length ({metadata_len}) is close to chunk size "
        f"({self.chunk_size}). Resulting chunks are less than 50 tokens. "
        "Consider increasing the chunk size or decreasing the size of "
        "your metadata to avoid this.",
        flush=True,
    )

📖 " if chunk_overlap > chunk_size:

📖 "SENTENCE_CHUNK_OVERLAP = 200

chunk_size: int = Field(
    default=DEFAULT_CHUNK_SIZE,
    description="The token chunk size for each chunk.",
    gt=0,
)
chunk_overlap: int = Field(
    default=SENTENCE_CHUNK_OVERLAP,
    description="The token overlap of each chunk when splitting.",
    ge=0,
)

📖 " # Update search position to start from next character after this node's START

📖 " if self.include_prev_next_rel:

node_content = node.get_content(metadata_mode=MetadataMode.NONE)
start_char_idx = parent_doc.text.find(node_content, search_start)

# update start/end char idx
if start_char_idx >= 0 and isinstance(node, TextNode):
    node.start_char_idx = start_char_idx
    node.end_char_idx = start_char_idx + len(node_content)

📖 "An ingestion pipeline that can be applied to data." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

📖 "If a vector store is provided, nodes with embeddings will be added to the vector store." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=20),
        OpenAIEmbedding(),
    ],
)

nodes = pipeline.run(documents=documents)

📖 "return sha256((nodes_str + transform_string).encode("utf-8")).hexdigest()" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

📖 " cached_nodes = cache.get(hash, collection=cache_collection)

# 第一次 run 会计算并缓存每一步;不改文档与配置再次 run,transformation 直接命中缓存
pipeline.run(documents=documents)          # 计算 + 写缓存
pipeline.run(documents=documents)          # 命中缓存,跳过重复转换
pipeline.persist("./pipeline_storage")     # 把 cache(与 docstore)落盘

📖 "Document de-duplication de-deduplication strategies work by comparing the hashes or ids stored in the document store." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

📖 " else:

from llama_index.core.ingestion import IngestionPipeline, DocstoreStrategy
from llama_index.core.storage.docstore import SimpleDocumentStore

pipeline = IngestionPipeline(
    transformations=[...],
    docstore=SimpleDocumentStore(),
    vector_store=vector_store,
    docstore_strategy=DocstoreStrategy.UPSERTS_AND_DELETE,  # 输入里删掉的文档也会被清出索引
)

📖 "to apply upsert/delete semantics; falling back to 'duplicates_only' for this run. " — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

📖 "pipeline.docstore_strategy is unchanged." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

# 只挂 docstore、没挂 vector_store,且 strategy=UPSERTS
pipeline = IngestionPipeline(
    transformations=[...],
    docstore=SimpleDocumentStore(),          # 无 vector_store
    docstore_strategy=DocstoreStrategy.UPSERTS,
)
pipeline.run(documents=docs)   # 触发 UserWarning,本次实际按 duplicates_only 执行

📖 "num_workers (Optional[int], optional): The number of parallel processes to use." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

📖 "External backends write through to shared storage and need no merge." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/ingestion/pipeline.py

# 用 4 个进程并行跑转换;超过 CPU 数会被自动压回上限
nodes = pipeline.run(documents=documents, num_workers=4)

📖 "Settings for the Llama Index, lazily initialized." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

📖 "Settings = _Settings()" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

📖 "self._node_parser = SentenceSplitter()" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

from llama_index.core import Settings

# 此刻 _embed_model 仍为 None,尚未构造任何模型
Settings.embed_model = my_embedding      # 触发 setter -> resolve_embed_model
size = Settings.chunk_size               # 走 node_parser 默认 SentenceSplitter

📖 "self._embed_model = resolve_embed_model("default")" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

📖 "self._embed_model.callback_manager = self._callback_manager" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

📖 "self._llm = resolve_llm("default")" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/settings.py

Settings.embed_model = "local"          # setter: resolve_embed_model("local")
Settings.callback_manager = my_manager  # 之后设置
emb = Settings.embed_model              # getter 再次把 my_manager 注入 emb

📖 "Get a list of text embeddings, with batching." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "if idx == len(texts) - 1 or len(cur_batch) == self.embed_batch_size:" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "The batch size for embedding calls." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

文本: [t0, t1, t2, t3, t4]
批1(满2条 flush): [t0, t1]
批2(满2条 flush): [t2, t3]
批3(到末尾 flush): [t4]

📖 "Cache for the embeddings: if None, the embeddings are not cached" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "Rate limiter instance to throttle API calls." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "Tuples of (index, text) to be able to keep same order of embeddings" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

cached_emb = self.embeddings_cache.get(key=text, collection="embeddings")
if cached_emb is not None:
    text_embedding = cached_emb[next(iter(cached_emb.keys()))]  # 命中,跳过模型
else:
    if self.rate_limiter is not None:
        self.rate_limiter.acquire()                            # 未命中才限流
    text_embedding = self._get_text_embedding(text)

📖 "DEFAULT = "cosine"" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "Using -euclidean distance as similarity to achieve same ranking order" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

📖 "Non-sensitive representation of this embedding model for observability." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/base/embeddings/base.py

# EUCLIDEAN: 取负距离,保证"越大越相似"
return -float(np.linalg.norm(np.array(embedding1) - np.array(embedding2)))
# DOT_PRODUCT
return np.dot(embedding1, embedding2)
# DEFAULT (cosine)
product = np.dot(embedding1, embedding2)
norm = np.linalg.norm(embedding1) * np.linalg.norm(embedding2)
return product / norm

📖 "Take a single step with the function calling agent." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "current_llm_input = [*llm_input, *scratchpad]" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "raise ValueError(\"LLM must be a FunctionCallingLLM\")" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

scratchpad: List[ChatMessage] = await ctx.store.get(
    self.scratchpad_key, default=[]
)
current_llm_input = [*llm_input, *scratchpad]
# ... 取 response ...
scratchpad.append(last_chat_response.message)
await ctx.store.set(self.scratchpad_key, scratchpad)

📖 "Handle tool call results for function calling agent." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "only add to scratchpad if we didn't select the handoff tool" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

if (
    tool_call_result.return_direct
    and tool_call_result.tool_name != "handoff"
):
    scratchpad.append(
        ChatMessage(
            role="assistant",
            content=str(tool_call_result.tool_output.content),
            additional_kwargs={"tool_call_id": tool_call_result.tool_id},
        )
    )
    break

📖 "Adds all in-progress messages to memory." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "await memory.aput_messages(scratchpad)" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

scratchpad: List[ChatMessage] = await ctx.store.get(
    self.scratchpad_key, default=[]
)
await memory.aput_messages(scratchpad)

# reset scratchpad
await ctx.store.set(self.scratchpad_key, [])

return output

📖 "If True, the agent will call multiple tools in parallel. If False, the agent will call tools sequentially." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "The tool to try and force to call on the first iteration of the agent." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

📖 "Only add tool choice if set and if its the first response" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/function_agent.py

# Only add tool choice if set and if its the first response
if (
    self.initial_tool_choice is not None
    and current_llm_input[-1].role == "user"
):
    chat_kwargs["tool_choice"] = self.initial_tool_choice

📖 "if num_iterations >= max_iterations:" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/base_agent.py

📖 "Method to handle max iterations. 'force' raises an error (default). 'generate' makes one final LLM call to generate a response." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/base_agent.py

flowchart TD
    A[init_run] --> B[setup_agent]
    B --> C[run_agent_step 调 take_step]
    C --> D[parse_agent_output]
    D -->|无 tool_calls| E[finalize + StopEvent]
    D -->|有 tool_calls| F[call_tool]
    F --> G[aggregate_tool_results]
    G -->|回灌 AgentInput| B
    D -->|num_iterations >= max| H{early_stopping_method}
    H -->|force| I[raise WorkflowRuntimeError]
    H -->|generate| J[生成收尾回答 StopEvent]

📖 "Handoff control of that chat to the given agent." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/multi_agent_workflow.py

📖 "Do not create a handoff tool if there is only one agent" — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/multi_agent_workflow.py

📖 "If the LLM is a function calling model, the workflow will use the FunctionAgent." — source: https://github.com/run-llama/llama_index/blob/main/llama-index-core/llama_index/core/agent/workflow/multi_agent_workflow.py

await ctx.store.set("next_agent", to_agent)
handoff_output_prompt = await ctx.store.get(
    "handoff_output_prompt", default=DEFAULT_HANDOFF_OUTPUT_PROMPT
)

return handoff_output_prompt.format(to_agent=to_agent, reason=reason)