垦荒学园
  • 中文

      • 实战案例

      三个由浅入深的例子：纯查询引擎、带工具的 FunctionAgent、以及 Context 多轮。LLM 用 gpt-4o-mini（历史默认曾是 gpt-3.5-turbo，以官方为准）。先 export OPENAI_API_KEY，并准备 data/。

      更偏 Agent harness 的写法对照 LangChain 实战；角色班组对照 CrewAI；概念对照 RAG。

      案例 1：目录问答 + 持久化

      目标： 对公司 data/ 做问答，第二次启动不再 embedding。

      from pathlib import Path
from llama_index.core import (
    Settings,
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    load_index_from_storage,
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

PERSIST = "./storage"
if Path(PERSIST).exists():
    index = load_index_from_storage(StorageContext.from_defaults(persist_dir=PERSIST))
else:
    documents = SimpleDirectoryReader("data", recursive=True).load_data()
    index = VectorStoreIndex.from_documents(documents)
    index.storage_context.persist(persist_dir=PERSIST)

qe = index.as_query_engine(similarity_top_k=4)
response = qe.query("加班审批流程是什么？请根据资料回答。")
print(response)
for n in response.source_nodes:
    print("-", n.metadata.get("file_name") or n.metadata.get("filename"), n.score)

      验证： 问文档里独有的专有名词；删掉 OPENAI_API_KEY 应失败；删 storage/ 应重新建索引。生产可把 storage/ 换成 Chroma。

      案例 2：检索工具 + 计算器的 FunctionAgent

      目标： 模型自己决定搜文档还是做乘法（官方 starter 的完整形态）。

      import asyncio
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

documents = SimpleDirectoryReader("data").load_data()
query_engine = VectorStoreIndex.from_documents(documents).as_query_engine()

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

async def search_documents(query: str) -> str:
    """Search local files for factual answers."""
    return str(await query_engine.aquery(query))

agent = FunctionAgent(
    tools=[multiply, search_documents],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="Use search_documents for facts in the corpus. Use multiply for arithmetic.",
)

async def main():
    print(await agent.run("手册里怎么申请年假？另外 12*8 是多少？"))

if __name__ == "__main__":
    asyncio.run(main())

      验证： 单问数学不应搜库；单问制度应调用 search_documents。可用 QueryEngineTool.from_defaults 代替手写 async 函数，见 查询引擎。

      案例 3：Context 记住用户名字

      目标： 同一会话里第二轮能回答「我叫什么」。

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

def echo_policy(topic: str) -> str:
    """Return a canned policy snippet for demos."""
    return f"[demo] Policy about {topic}: submit via HR portal."

agent = FunctionAgent(
    tools=[echo_policy],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="Remember the user's name across turns. Use tools for policy.",
)

async def main():
    ctx = Context(agent)
    await agent.run("我叫 Alex，请记住。", ctx=ctx)
    r = await agent.run("我叫什么？年假政策呢？", ctx=ctx)
    print(r)

if __name__ == "__main__":
    asyncio.run(main())

      验证： 不传 ctx 的第二轮不应知道名字。每个终端用户使用独立 Context（或官方文档中的持久化方式），不要全站共用一个。

      需要固定「检索 → 质检 → 重写」循环时，用 Workflows 的 @step，不要把所有控制流塞进一个巨大 system prompt。

      反模式

      反模式改进
      每次启动 from_documentspersist 或 from_vector_store
      不看 source_nodes 就调 Prompt先确认金标准段落是否被召回
      全公司一个 collection 无租户字段分库或 metadata filter
      把 LlamaIndex 当唯一后端事务库业务数据仍在 SQL；这里只选上下文
      默认模型不写死显式 gpt-4o-mini（或你的生产型号）+ 同一 embedding

      下一步

      评论