LangGraph 與 LangFuse：打造智能觀測系統的完美組合

7 min readAug 16, 2024

🚀 想要打造下一代 AI 應用，卻被複雜性和可觀測性難題困擾？
揭秘 LangGraph 和 LangFuse：AI 開發的革命性組合！這篇文章將帶您深入探索如何:
✅ 輕鬆構建多智能體系統
✅ 實現毫秒級監控和分析
✅ 大幅提升 AI 專案的可靠性和效率

立即閱讀，開啟您的 AI 開發新篇章！

簡介

在人工智能快速發展的今天，如何有效地構建、監控和優化複雜的 AI 應用已成為一個關鍵挑戰。本文將為您介紹兩個強大的工具：LangGraph 和 LangFuse，以及它們如何協同工作，為您的 AI 專案帶來革命性的變革。

LangGraph：多智能體應用的革命性框架

LangGraph 是由 LangChain 團隊精心打造的開源框架，旨在簡化多智能體系統的構建過程。它的主要特點包括：

輕鬆構建具有狀態管理的多智能體系統
內建持久化功能，便於錯誤恢復
支援人機協作工作流

LangFuse：您的 AI 智能儀表板

如果說 LangGraph 是您的 AI 引擎，那麼 LangFuse 就是您的智能儀表板。通過 LangFuse 的 LangChain 集成，您可以：

自動追蹤 LangGraph 應用的每一個細節
深入分析複雜的多智能體互動
添加關鍵指標（如用戶反饋）
高效管理 LangGraph 中使用的提示詞

實戰：構建智能客服聊天機器人

讓我們通過一個簡單但強大的例子來展示 LangGraph 和 LangFuse 的協同效果。

步驟 1：構建基礎聊天機器人

首先，我們使用 LangGraph 的 `StateGraph` 來定義聊天機器人的結構：

from typing import Annotated
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.set_entry_point("chatbot")
graph_builder.set_finish_point("chatbot")

graph = graph_builder.compile()

這段代碼定義了一個簡單的聊天機器人結構，包括狀態管理和消息處理。

步驟 2：集成 LangFuse 觀測

接下來，我們添加 LangFuse 的 callbackHandler 來追蹤每一步操作：

from langfuse.callback import CallbackHandler

langfuse_handler = CallbackHandler()

for s in graph.stream({"messages": [HumanMessage(content="What is Langfuse?")]},
                      config={"callbacks": [langfuse_handler]}):
    print(s)

通過添加 LangFuse callbackHandler，我們能夠詳細記錄聊天機器人的每一次交互。

進階：多 Agent 超級團隊

讓我們再進一步，構建一個由多個專門 AI 智能體組成的超級團隊。

團隊成員設計

1. 研究員(Researcher)：負責搜索維基百科
2. 時間管理員(CurrentTime)：負責報告當前時間
3. 主管(supervisor)：負責任務分配

核心代碼展示

workflow = StateGraph(AgentState)

workflow.add_node("Researcher", research_node)
workflow.add_node("CurrentTime", currenttime_node)
workflow.add_node("supervisor", supervisor_chain)

for member in members:
    workflow.add_edge(member, "supervisor")

conditional_map = {k: k for k in members}
conditional_map["FINISH"] = END
workflow.add_conditional_edges("supervisor", lambda x: x["next"], conditional_map)

graph_2 = workflow.compile()

這段程式碼展示了如何構建一個多智能體工作流，包括節點添加、邊緣連接和條件路由。

LangFuse 的高級功能

評分系統

LangFuse 提供了靈活的評分功能，可以應用於單獨的觀察結果或整個追蹤過程：

@observe()
def research_agent(user_message):
    lf_handler = langfuse_context.get_current_langchain_handler()
    response = graph_2.invoke({"messages": [HumanMessage(content=user_message)]},
                        config={"callbacks": [lf_handler]})
    langfuse_context.score_current_trace(
        name="user-explicit-feedback",
        value=1,
        comment="The time is correct!"
    )
    return response

這個例子展示了如何使用 `@observe()` 裝飾器自動捕獲豐富的上下文信息，並如何為整個追蹤添加評分。

提示詞管理

LangFuse 的提示詞管理功能就像是一個專為 LLM 提示語設計的內容管理系統（CMS）：

langfuse.create_prompt(
    name="translator_system-prompt",
    prompt="你是一位翻譯專家,負責將所有輸入文本翻譯成日語。",
    labels=["production"]
)

langfuse_system_prompt = langfuse.get_prompt("translator_system-prompt")
langchain_system_prompt = langfuse_system_prompt.get_langchain_prompt()

這段代碼展示了如何創建、存儲和檢索提示詞，實現版本控制和輕鬆更新。