LangChain vs LlamaIndex vs CrewAI：AI 框架三巨头对比

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# 加载文档
loader = PyPDFLoader("report.pdf")
docs = loader.load()

# 分割
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = splitter.split_documents(docs)

# 向量化
vectorstore = Chroma.from_documents(splits, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 构建 RAG 链
system_prompt = (
    "你是一个助手。根据以下检索到的上下文回答问题：\n\n{context}"
)
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
])
question_answer_chain = create_stuff_documents_chain(ChatOpenAI(model="gpt-4o"), prompt)
rag_chain = create_retrieval_chain(retriever, question_answer_chain)

# 查询
response = rag_chain.invoke({"input": "这份报告的核心结论是什么？"})
print(response["answer"])

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 全局配置
Settings.llm = OpenAI(model="gpt-4o")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 加载文档——自动分词、分块
documents = SimpleDirectoryReader(input_files=["report.pdf"]).load_data()

# 构建索引
index = VectorStoreIndex.from_documents(documents)

# 创建查询引擎
query_engine = index.as_query_engine(
    similarity_top_k=4,
    response_mode="compact",
)

# 查询
response = query_engine.query("这份报告的核心结论是什么？")
print(response)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileWriterTool

search_tool = SerperDevTool()
writer_tool = FileWriterTool()

# 定义研究员 Agent
researcher = Agent(
    role="资深研究员",
    goal="深入调研 AI Agent 市场现状",
    backstory="你是一位拥有 10 年经验的科技行业分析师，擅长从海量信息中提炼关键洞察。",
    tools=[search_tool],
    llm="gpt-4o",
    verbose=True,
)

# 定义撰稿人 Agent
writer = Agent(
    role="技术撰稿人",
    goal="将研究成果转化为高质量文章",
    backstory="你是顶级科技媒体的技术编辑，擅长将复杂技术概念转化为通俗文章。",
    tools=[writer_tool],
    llm="gpt-4o",
    verbose=True,
)

# 定义任务
research_task = Task(
    description="调研 2024-2026 年 AI Agent 市场的主要玩家、技术趋势和商业模式。",
    expected_output="一份 2000 字的研究报告，包含市场数据、技术分析和趋势预测。",
    agent=researcher,
)

write_task = Task(
    description="基于研究报告撰写一篇面向开发者的技术博客文章。",
    expected_output="一篇 3000 字的 Markdown 格式博客文章。",
    agent=writer,
    context=[research_task],
)

# 组建 Crew 并执行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]
    research_output: str
    article_output: str

llm = ChatOpenAI(model="gpt-4o")

def research_node(state: State):
    prompt = [
        SystemMessage(content="你是资深研究员，深入调研 AI Agent 市场。"),
        HumanMessage(content="调研 2024-2026 年 AI Agent 市场，输出 2000 字报告。"),
    ]
    response = llm.invoke(prompt)
    return {"research_output": response.content}

def writer_node(state: State):
    prompt = [
        SystemMessage(content="你是技术撰稿人，将研究内容转化为博客文章。"),
        HumanMessage(content=f"基于以下研究内容写一篇 3000 字博客：\n\n{state['research_output']}"),
    ]
    response = llm.invoke(prompt)
    return {"article_output": response.content}

# 构建图
workflow = StateGraph(State)
workflow.add_node("researcher", research_node)
workflow.add_node("writer", writer_node)
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", END)
workflow.set_entry_point("researcher")

app = workflow.compile()
result = app.invoke({"messages": []})
print(result["article_output"])

用户输入 → LlamaIndex（RAG 检索）
         → CrewAI（多 Agent 处理）
         → LangGraph（状态管理与流程控制）
         → 输出

from llama_index.core import VectorStoreIndex
from crewai import Agent, Task, Crew
from crewai.tools import tool

# 用 LlamaIndex 构建 RAG 工具
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=5)

@tool("Knowledge Base Search")
def search_knowledge(query: str) -> str:
    """搜索知识库获取相关信息"""
    return str(query_engine.query(query))

# 将 RAG 工具提供给 CrewAI Agent
analyst = Agent(
    role="数据分析师",
    goal="基于知识库回答问题",
    backstory="你是专业分析师。",
    tools=[search_knowledge],
)

task = Task(
    description="分析知识库中的数据趋势",
    expected_output="分析报告",
    agent=analyst,
)

crew = Crew(agents=[analyst], tasks=[task])
result = crew.kickoff()