大模型开发必备!LangChain、LlamaIndex等7大框架一次性说清楚

在大模型时代,开发者们面临着海量的数据处理、复杂的任务编排以及多样化的应用场景等诸多挑战。而各种大模型开发框架应运而生,它们如同一把把神奇的钥匙,能够解锁大模型的强大能力,让开发者们更加高效地构建各种智能应用。然而,面对 LangChain、LlamaIndex、LangGraph、Agently、LangFuse、LangSmith、LangFlow 这么多框架,它们到底有何不同?又该如何选择呢?别急,这篇文章将一次性为你说清楚,让你在大模型开发的道路上不再迷茫,轻松找到最适合你的那把“钥匙”!

1. 框架概览与分类

1.1 框架概览

这些框架都是围绕大语言模型(LLM)构建的,旨在简化开发流程,提高开发效率。它们各自有着独特的设计理念和功能特点,适用于不同的开发场景。

1.2 框架分类

我们可以将这些框架大致分为以下几类:

  • 通用开发框架:如 LangChain、LangFlow,它们提供了丰富的组件和灵活的架构,能够满足各种复杂的开发需求。
  • 检索增强框架:如 LlamaIndex,专注于通过检索增强生成(RAG)的方式,将大模型与外部数据源高效结合,提升问答系统的性能。
  • 图结构框架:如 LangGraph,基于图结构优化代理逻辑流程,适合构建多步骤任务处理系统。
  • 代理框架:如 Agently,专注于构建智能代理系统,能够快速实现聊天机器人、任务调度器等应用。
  • 可观测性和分析平台:Langfuse 是一个开源的可观测性和分析平台,专为由大型语言模型(LLM)驱动的应用而设计。我们的使命是帮助开发人员和组织构建并改进 LLM 应用程序。为此,我们通过先进的跟踪和分析模块深入了解模型的成本、质量和延迟。
  • 开发辅助平台:如 LangSmith,主要提供调试、测试、评估和监控等功能,帮助开发者优化开发流程。

2. 框架对比

  • 七大开发框架对比
  • 两大流程开发框架对比

3. 具体框架介绍与开发案例

3.1 LangChain

核心特点:LangChain 是一个高度模块化的框架,提供了丰富的组件(如提示工程、索引、记忆等),并通过链式架构将这些组件组合成复杂的处理流程。它支持多种任务类型,包括文档问答、聊天机器人、任务自动化等。

开发案例

from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableLambda
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser
from dotenv import load_dotenv, find_dotenv

# 加载环境变量
_ = load_dotenv(find_dotenv())

# 初始化语言模型
llm = ChatOpenAI(model="qwen2.5:7b", temperature=0)


# 定义获取会话历史的方法
def get_session_history(session_id):
    # 使用最新的API或参数替换下面的数据库连接方式
    return SQLChatMessageHistory(connection="sqlite:///memory.db", session_id=session_id)


# 创建一个简单的Runnable,它接受消息并调用语言模型的invoke方法
runnable = RunnableLambda(lambda x: llm.invoke(x)) | StrOutputParser()

# 将历史记录功能添加到Runnable中
runnable_with_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
)

# 使用特定的会话ID调用runnable_with_history
response = runnable_with_history.invoke(
    [HumanMessage(content="你好,我叫王XX")],
    config={"session_id": "wzr"},  # 确保这里传递的是`session_id`
)

print(response)

3.2 LlamaIndex

核心特点:LlamaIndex 是一个专注于检索增强生成(RAG)的框架,通过向量索引和检索技术,能够快速匹配并返回最相关的信息块,为大模型提供精准的上下文支持。

开发案例

from llama_index import SimpleDirectoryReader, GPTVectorStoreIndex, ServiceContext
from llama_index import OpenAIEmbedding

# 加载文档
documents = SimpleDirectoryReader("path/to/your/text_files").load_data()

# 构建索引
embedding_model = OpenAIEmbedding()
service_context = ServiceContext.from_defaults(embeddings=embedding_model)
index = GPTVectorStoreIndex.from_documents(documents, service_context=service_context)

# 创建查询引擎
query_engine = index.as_query_engine()

# 查询
response = query_engine.query("请简要介绍文档中的主要内容。")
print(response)

3.3 LangGraph

核心特点:LangGraph 基于图结构优化代理逻辑流程,通过节点、边和条件边的概念,能够清晰地表示复杂的执行路径和逻辑关系。它支持异步执行,提高系统响应速度和效率。

开发案例

from langgraph import Graph, Node, Edge

# 创建图
graph = Graph()

# 添加节点
node1 = Node("任务1")
node2 = Node("任务2")
graph.add_node(node1)
graph.add_node(node2)

# 添加边
edge = Edge(node1, node2, condition="完成任务1")
graph.add_edge(edge)

# 执行图
result = graph.run()
print(result)

3.4 Agently

核心特点:Agently 是一个专注于构建智能代理系统的框架,提供简洁强大的工具来定义和管理代理行为。它内置丰富的功能和插件,能够快速实现聊天机器人、任务调度器等应用。

开发案例

"""创建Agent实例和Workflow实例"""
import Agently
agent = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", "https://api.moonshot.cn/v1")
        .set_settings("model.OAIClient.options", { "model": "moonshot-v1-8k" })
        .set_settings("model.OAIClient.auth", { "api_key": "********" })
)
workflow = Agently.Workflow()
"""创建执行块(Chunk)"""
# 用户输入块
@workflow.chunk()
def user_input(inputs, storage):
    return input("[User]: ")

# Agent回复块
@workflow.chunk()
def assistant_reply(inputs, storage):
    chat_history = storage.get("chat_history") or []
    reply = (
        agent
            .chat_history(chat_history)
            .input(inputs["default"])
            .start()
    )
    print("[Assistant]: ", reply)
    return reply

# 对话记录更新块
@workflow.chunk()
def update_chat_history(inputs, storage):
    chat_history = storage.get("chat_history", [])
    chat_history.append({ "role": "user", "content": inputs["user_input"] })
    chat_history.append({ "role": "assistant", "content": inputs["assistant_reply"] })
    storage.set("chat_history", chat_history)
    return

# 道别块
@workflow.chunk()
def goodbye(inputs, storage):
    print("Bye~")
    return

# 取出对话记录块
@workflow.chunk()
def get_all_chat_history(inputs, storage):
    return storage.get("chat_history")

"""连接执行块"""
workflow.connect_to("user_input")
(
    workflow.chunks["user_input"]
        .if_condition(lambda return_value, storage: return_value == "#exit")
            .connect_to("goodbye")
            .connect_to("get_all_chat_history")
            .connect_to("end")
        .else_condition()
            .connect_to("assistant_reply")
)
workflow.chunks["user_input"].connect_to("update_chat_history.user_input")
workflow.chunks["assistant_reply"].connect_to("update_chat_history.assistant_reply")
workflow.chunks["update_chat_history"].connect_to("user_input")

"""获取工作流Mermaid代码(可绘图)"""
print(workflow.draw())

"""启动工作流"""
"""将需要输出的数据连接到`end`块即可获得返回结果"""
chat_history = workflow.start()
print(chat_history["default"])

3.5 LangFuse

核心特点:LangFuse 是一个多语言框架,提供强大的多语言支持,能够在一个框架内处理多种语言的数据和任务。它适用于多语言文本生成、翻译、情感分析等国际化的应用场景。

开发案例

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    AIMessage,  # 等价于OpenAI接口中的assistant role
    HumanMessage,  # 等价于OpenAI接口中的user role
    SystemMessage  # 等价于OpenAI接口中的system role
)
from datetime import datetime
from langfuse.decorators import langfuse_context, observe

now = datetime.now()

llm = ChatOpenAI()

messages = [
    SystemMessage(content="你是AGIClass的课程助理。"),
]

session_id = "chat-"+now.strftime("%d/%m/%Y %H:%M:%S")

@observe()
def chat_one_turn(user_input, user_id, turn_id):
    langfuse_context.update_current_trace(
        name=f"ChatTurn{turn_id}",
        user_id=user_id,
        session_id=session_id
    )
    langfuse_handler = langfuse_context.get_current_langchain_handler()
    messages.append(HumanMessage(content=user_input))
    response = llm.invoke(messages, config={"callbacks": [langfuse_handler]})
    messages.append(response)
    return response.content    

user_id="test"
turn_id = 0
while True:
    user_input = input("User: ")
    if user_input.strip() == "":
        break
    reply = chat_one_turn(user_input, user_id, turn_id)
    print("AI: "+reply)
    turn_id += 1
    langfuse_context.flush()

3.6 LangSmith

核心特点:LangSmith 是一个开发辅助平台,提供调试、测试、评估和监控等功能,帮助开发者在开发过程中更高效地进行问题定位、性能优化和质量保证。

开发案例

from langchain.prompts import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough

model = ChatOpenAI(model="gpt-3.5-turbo")

prompt = ChatPromptTemplate.from_messages([
    HumanMessagePromptTemplate.from_template("Say hello to {input}!")
])


# 定义输出解析器
parser = StrOutputParser()

chain = (
    {"input": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

chain.invoke("王祖蓝")

3.7 LangFlow

核心特点:LangFlow 是一个流程化开发框架,提供可视化界面,使得开发者能够以直观的方式定义应用的逻辑结构,无需编写大量代码。它支持流程的版本管理和协作开发,方便团队进行项目管理和协作。

开发案例

docker run -it --rm \
    -p 7860:7860 \
    --env-file .env \
    langflowai/langflow:latest

4. 如何选择适合你的框架

选择合适的框架需要根据你的具体需求和开发场景来决定:

  • 如果你需要构建复杂的任务处理系统,推荐使用 LangChainLangGraph
  • 如果你的应用场景主要涉及文档问答或知识库检索,LlamaIndex 是一个不错的选择。
  • 对于需要快速实现聊天机器人或任务调度器的场景,Agently 可以大大简化开发流程。
  • 如果你的项目涉及多语言处理,LangFuse 将是你的得力助手。
  • 在开发过程中,如果你需要强大的调试、测试和监控功能,LangSmith 将为你提供有力支持。
  • 如果你是一个团队开发者,需要进行协作开发和流程管理,LangFlow 将是一个理想的选择。
原文链接:,转发请注明来源!