整合 Langfuse 至 IPFLY 驅動的 AI 代理以實現企業可觀測性

13次閱讀

Langfuse 是一款開源 LLM 工程平台，為 AI 代理提供可觀測性、追蹤與監控功能——這對於企業級使用案例（如合規追蹤）至關重要，因為在此類場景中，可靠性與透明度是不可妥協的。當使用 LangChain 建構聚焦於合規的 AI 代理時，最大的挑戰在於不受限制地存取權威網路資料（例如：法規更新、政府指南）。

IPFLY 的優質代理解決方案（遍佈 190 多個國家的 9000 多萬個全球 IP、靜態/動態住宅代理及資料中心代理）解決了這個問題：多層 IP 過濾可繞過反爬蟲措施，全球覆蓋範圍解鎖區域特定的合規資料，而 99.9% 的正常運作時間確保資料擷取的一致性。本指南將逐步引導您建構 LangChain 合規 AI 代理、整合 IPFLY 進行網路資料收集，並使用 Langfuse 追蹤從提示輸入到代理驅動的網路爬蟲的每一個步驟。

整合 Langfuse 至 IPFLY 驅動的 AI 代理以實現企業可觀測性

Langfuse、AI 代理可觀測性與 IPFLY 角色介紹

企業級 AI 代理（尤其是合規追蹤類）依賴兩大支柱：準確的網路資料（以掌握 GDPR、CCPA 等最新法規）和完整的可觀測性（以驗證決策、追蹤成本並確保合規）。

Langfuse：為 LLM 應用程式提供端對端的追蹤、指標與除錯功能——讓團隊能監控 AI 代理工作流程的每個步驟（提示、工具呼叫、回應）。

LangChain：協調 AI 代理邏輯，將 LLM 連接到外部工具（如網路爬蟲）以擷取資料。

IPFLY：透過專為 AI 設計的代理服務消除網路資料存取瓶頸：動態住宅代理模仿真實使用者以避免被封鎖，靜態住宅代理確保持續存取可信賴的法規網站，資料中心代理則處理大規模爬蟲——所有服務都具備全球覆蓋能力，可取得區域合規資料。

這些工具共同打造了企業級技術堆疊：IPFLY 為代理提供高品質網路資料，LangChain 管理工作流程，而 Langfuse 確保對效能與可靠性的完整可見性。

為何將 Langfuse 整合到您的 AI 代理中

用於合規的 AI 代理會與敏感文件、外部網路資料和複雜的法規規則互動——這些環節的盲點可能導致代價高昂的錯誤或不合規。Langfuse 透過以下方式解決此問題：

  • 提供端對端追蹤：監控每個工具呼叫（例如 IPFLY 網路爬蟲）與資料來源以驗證洞察。
  • 追蹤關鍵指標：測量延遲、LLM 成本和網路爬蟲成功率（對最佳化 IPFLY 代理使用至關重要）。
  • 實現快速除錯：透過詳細記錄識別失敗的爬蟲、過時的提示或 LLM 幻覺。
  • 支援合規性：為稽核建立代理行為的不可變記錄。

當與 IPFLY 配對時，Langfuse 不僅確保您的代理正常運作——還能證明它以可靠且合法的方式運作。

如何使用 Langfuse 追蹤合規性 AI 代理（LangChain + IPFLY）

我們將建構一個企業級合規 AI 代理，它能夠：

  1. 載入內部 PDF 文件（例如：資料處理工作流程）。
  2. 分析 PDF 以識別隱私/法規風險。
  3. 使用 IPFLY 代理搜尋最新法規（SERP 資料）並爬蟲權威來源（政府網站）。
  4. 生成包含內部文件和網路資料引用的合規報告。
  5. 整合 Langfuse 以實現完整工作流程追蹤。

先決條件

開始之前，請確保您擁有：

  • Python 3.10 或更高版本
  • OpenAI API 金鑰（或其他 LLM 提供者 API 金鑰）
  • IPFLY 帳戶（含 API 金鑰和動態住宅代理存取權限）
  • Langfuse 帳戶（已設定公開/私密 API 金鑰）
  • 對 LangChain 和 Python 的基本熟悉度

步驟 #1：設定您的 LangChain AI 代理專案

建立專案資料夾和虛擬環境：

mkdir compliance-ai-agent-ipfly-langfuse
cd compliance-ai-agent-ipfly-langfuse
python -m venv .venv
# 啟用：macOS/Linux → source .venv/bin/activate；Windows → .venv\Scripts\activate
pip install langchain langchain-openai langgraph langchain-community pypdf python-dotenv langfuse requests

建立兩個檔案：agent.py（核心邏輯）和 .env（憑證）：

compliance-ai-agent-ipfly-langfuse/
├── .venv/
├── agent.py
└── .env

步驟 #2：設定環境變數讀取

在 agent.py 中，載入環境變數以安全地儲存敏感憑證：

Python

复制

from dotenv import load_dotenv
load_dotenv()  # 從 .env 檔案載入變數

將憑證新增至您的 .env 檔案（我們將在後續步驟中填入 IPFLY/Langfuse/OpenAI 金鑰）：

env

OPENAI_API_KEY="<YOUR_OPENAI_API_KEY>"
IPFLY_API_KEY="<YOUR_IPFLY_API_KEY>"
IPFLY_PROXY_ENDPOINT="http://[USERNAME]:[PASSWORD]@proxy.ipfly.com:8080"
LANGFUSE_SECRET_KEY="<YOUR_LANGFUSE_SECRET_KEY>"
LANGFUSE_PUBLIC_KEY="<YOUR_LANGFUSE_PUBLIC_KEY>"
LANGFUSE_BASE_URL="<YOUR_LANGFUSE_BASE_URL>"

步驟 #3：準備您的 IPFLY 帳戶

IPFLY 為代理的網路資料收集提供動力（SERP 搜尋 + 法規網站爬蟲）。設定方式如下：

  1. 登入您的 IPFLY 帳戶並產生 API 金鑰（位於「帳戶設定」下）。
  2. 記下您的代理端點（在 IPFLY 儀表板中提供）——包含您的使用者名稱、密碼和連接埠。
  3. 對於合規使用案例，選擇動態住宅代理（以避免在政府/法規網站被封鎖）或靜態住宅代理（以持續存取可信賴來源）。

IPFLY 在此代理中的關鍵優勢：

  • 9000 多萬個真實使用者 IP：模仿真人瀏覽以繞過反爬蟲工具（例如在 GDPR.eu 上的驗證碼）。
  • 190 多個國家覆蓋範圍：爬蟲多國法規（例如：CCPA 適用於美國、PIPEDA 適用於加拿大）。
  • 多層 IP 過濾：確保不使用黑名單 IP，維護資料收集規則的合規性。
  • 99.9% 正常運作時間：保證持續存取關鍵法規資料。

步驟 #4：為 LangChain 建構 IPFLY 工具

建立自訂 LangChain 工具以使用 IPFLY 代理處理 SERP 搜尋和網路爬蟲。將以下內容新增至 agent.py：

Python

import requests
from bs4 import BeautifulSoup
from langchain.tools import Tool

class IPFLYSERPTool(Tool):
    """使用 IPFLY 代理檢索 SERP 資料以進行法規搜尋查詢的工具。"""

    def __init__(self):
        super().__init__(
            name="ipfly_serp_search",
            description="使用 IPFLY 代理在 Google 上搜尋法規關鍵字（例如 'GDPR data retention'）。回傳前 5 個搜尋結果（優先政府網站）。",
            func=self.run
        )
        self.proxy = os.getenv("IPFLY_PROXY_ENDPOINT")

    def run(self, query: str) -> str:
        """使用 IPFLY 代理執行 SERP 搜尋。"""
        params = {"q": query, "hl": "en", "gl": "us"}  # 可針對區域法規自訂（例如 "eu" 用於 GDPR）
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        try:
            response = requests.get("https://www.google.com/search",
                params=params,
                proxies={"http": self.proxy, "https": self.proxy},
                headers=headers,
                timeout=30)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")
            results = []
            
            # 提取前 5 個自然搜尋結果（優先政府網站）
            for g in soup.find_all("div", class_="g")[:5]:
                title = g.find("h3").get_text(strip=True) if g.find("h3") else None
                url = g.find("a")["href"] if g.find("a") else None
                if title and url and ("gov" in url or "regulatory" in url):
                    results.append({"title": title, "url": url})
            
            return json.dumps(results, indent=2)
        except Exception as e:
            return f"SERP 搜尋失敗：{str(e)}"

class IPFLYWebScraperTool(Tool):
    """使用 IPFLY 代理爬蟲法規網站的工具（回傳 Markdown 格式內容）。"""

    def __init__(self):
        super().__init__(
            name="ipfly_web_scraper",
            description="使用 IPFLY 代理爬蟲法規網站（例如政府網站）。回傳乾淨的 Markdown 格式文字，供 LLM 分析。",
            func=self.run
        )
        self.proxy = os.getenv("IPFLY_PROXY_ENDPOINT")

    def run(self, url: str) -> str:
        """使用 IPFLY 代理爬蟲網頁。"""
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        try:
            response = requests.get(
                url,
                proxies={"http": self.proxy, "https": self.proxy},
                headers=headers,
                timeout=30)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")
            
            # 提取主要內容（移除廣告/導覽列）
            for script in soup(["script", "style", "nav", "aside", "footer"]):
                script.decompose()
            
            text = soup.get_text(strip=True, separator="\n")
            lines = [line.strip() for line in text.split("\n") if line.strip()]
            markdown = "\n\n".join(lines[:50])  # 限制為 50 行以符合 LLM 上下文
            
            return f"來源：{url}\n\n{markdown}"
        except Exception as e:
            return f"網路爬蟲失敗：{str(e)}"

# 初始化 IPFLY 工具
ipfly_serp_tool = IPFLYSERPTool()
ipfly_scraper_tool = IPFLYWebScraperTool()

步驟 #5：整合 LLM

將 OpenAI（或您偏好的 LLM）新增至 agent.py 以為代理提供分析能力：

Python

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-5-mini",  # 請替換為您的 LLM（例如 gpt-4o）
    api_key=os.getenv("OPENAI_API_KEY")
)

步驟 #6：定義合規性 AI 代理

結合 LLM、IPFLY 工具和系統提示來建立代理。將以下內容新增至 agent.py：

Python

from langchain.agents import create_agent
from langchain_core.prompts import PromptTemplate

# 用於合規追蹤的系統提示
system_prompt = """
您是合規追蹤專家。您的角色是分析內部文件的隱私/法規風險，並使用 IPFLY 爬蟲取得的即時網路資料進行驗證。
請遵循以下規則：
1. 分析輸入的 PDF 以識別關鍵法規層面（例如資料保留、刪除）。
2. 產生 2-3 個簡潔的 SERP 查詢（最多 5 個字詞）以尋找更新後的法規。
3. 使用 ipfly_serp_search 取得頂級法規網站（優先政府來源）。
4. 使用 ipfly_web_scraper 從這些網站提取內容。
5. 建立包含以下內容的報告：
   - 來自內部 PDF 的引用。
   - 來自爬蟲網路資料的洞察。
   - 清晰的合規建議。
6. 僅使用 IPFLY 爬蟲取得的來源和輸入的 PDF——絕不捏造資訊。
"""

# 工具清單（IPFLY + LLM）
tools = [ipfly_serp_tool, ipfly_scraper_tool]

# 建立代理（由 LangGraph 驅動）
agent = create_agent(
    llm=llm,
    tools=tools,
    system_prompt=system_prompt
)

步驟 #7：載入 PDF 並建立提示

新增邏輯以載入內部 PDF 文件並為代理生成提示。將以下內容新增至 agent.py：

Python

from langchain_community.document_loaders import PyPDFDirectoryLoader

# 建立 PDF 輸入資料夾
os.makedirs("./input", exist_ok=True)

# 載入 PDF 文件
loader = PyPDFDirectoryLoader("./input")
docs = loader.load()
internal_doc = "\n\n".join([doc.page_content for doc in docs])

# 代理的提示模板
prompt_template = PromptTemplate.from_template("""
分析以下內部文件的合規風險，並使用即時網路資料進行驗證：

PDF 內容：
{pdf}

生成包含 PDF 引用和爬蟲法規洞察的簡潔合規報告。
""")

# 建立最終提示
prompt = prompt_template.format(pdf=internal_doc)

步驟 #8：設定 Langfuse 以實現可觀測性

  1. 建立 Langfuse 帳戶（提供免費方案）並前往「專案設定」→「API 金鑰」。
  2. 產生公開/私密金鑰並將其新增至您的 .env 檔案（如步驟 #2 所示）。

步驟 #9：整合 Langfuse 追蹤

為代理新增 Langfuse 追蹤功能，以監控每一步驟（工具呼叫、LLM 輸出、延遲）。更新 agent.py：

Python

from langfuse import get_client
from langfuse.langchain import CallbackHandler

# 初始化 Langfuse 客戶端
langfuse = get_client(
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    base_url=os.getenv("LANGFUSE_BASE_URL")
)

# 建立 Langfuse 回呼處理器
langfuse_handler = CallbackHandler()

步驟 #10：最終程式碼

您完整的 agent.py 檔案將如下所示：

Python

import os
import json
import requests
from dotenv import load_dotenv
from bs4 import BeautifulSoup
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_core.prompts import PromptTemplate
from langchain_community.document_loaders import PyPDFDirectoryLoader
from langfuse import get_client
from langfuse.langchain import CallbackHandler

# 載入環境變數
load_dotenv()

# ------------------------------
# Langfuse 設定
# ------------------------------
langfuse = get_client(
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    base_url=os.getenv("LANGFUSE_BASE_URL")
)
langfuse_handler = CallbackHandler()

# ------------------------------
# 為 LangChain 建構 IPFLY 工具
# ------------------------------
class IPFLYSERPTool(Tool):
    def __init__(self):
        super().__init__(
            name="ipfly_serp_search",
            description="使用 IPFLY 代理搜尋 Google 法規關鍵字。回傳前 5 個政府/法規網站。",
            func=self.run
        )
        self.proxy = os.getenv("IPFLY_PROXY_ENDPOINT")

    def run(self, query: str) -> str:
        params = {"q": query, "hl": "en", "gl": "us"}
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        try:
            response = requests.get("https://www.google.com/search",
                params=params,
                proxies={"http": self.proxy, "https": self.proxy},
                headers=headers,
                timeout=30)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")
            results = []
            
            for g in soup.find_all("div", class_="g")[:5]:
                title = g.find("h3").get_text(strip=True) if g.find("h3") else None
                url = g.find("a")["href"] if g.find("a") else None
                if title and url and ("gov" in url or "regulatory" in url):
                    results.append({"title": title, "url": url})
            
            return json.dumps(results, indent=2)
        except Exception as e:
            return f"SERP 搜尋失敗：{str(e)}"

class IPFLYWebScraperTool(Tool):
    def __init__(self):
        super().__init__(
            name="ipfly_web_scraper",
            description="使用 IPFLY 代理爬蟲法規網站。回傳 Markdown 格式的內容。",
            func=self.run
        )
        self.proxy = os.getenv("IPFLY_PROXY_ENDPOINT")

    def run(self, url: str) -> str:
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        try:
            response = requests.get(
                url,
                proxies={"http": self.proxy, "https": self.proxy},
                headers=headers,
                timeout=30)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, "html.parser")
            
            for script in soup(["script", "style", "nav", "aside", "footer"]):
                script.decompose()
            
            text = soup.get_text(strip=True, separator="\n")
            lines = [line.strip() for line in text.split("\n") if line.strip()]
            markdown = "\n\n".join(lines[:50])
            
            return f"來源：{url}\n\n{markdown}"
        except Exception as e:
            return f"網路爬蟲失敗：{str(e)}"

# 初始化 IPFLY 工具
ipfly_serp_tool = IPFLYSERPTool()
ipfly_scraper_tool = IPFLYWebScraperTool()

# ------------------------------
# LLM 整合
# ------------------------------
llm = ChatOpenAI(
    model="gpt-5-mini",
    api_key=os.getenv("OPENAI_API_KEY")
)

# ------------------------------
# 合規性 AI 代理定義
# ------------------------------
system_prompt = """
您是合規追蹤專家。分析內部 PDF 的隱私/法規風險，並使用 IPFLY 爬蟲取得的即時網路資料進行驗證。
1. 從 PDF 中識別關鍵隱私/法規層面。
2. 產生 2-3 個簡潔的 SERP 查詢（最多 5 個字詞）。
3. 使用 ipfly_serp_search 尋找頂級政府/法規網站。
4. 使用 ipfly_web_scraper 從這些網站提取內容。
5. 建立包含 PDF 引用、網路洞察和合規建議的報告。
僅使用 PDF 和 IPFLY 爬蟲取得的資料——不捏造資訊。
"""

tools = [ipfly_serp_tool, ipfly_scraper_tool]
agent = create_agent(llm=llm, tools=tools, system_prompt=system_prompt)

# ------------------------------
# 載入 PDF 並建立提示
# ------------------------------
os.makedirs("./input", exist_ok=True)
loader = PyPDFDirectoryLoader("./input")
docs = loader.load()
internal_doc = "\n\n".join([doc.page_content for doc in docs])

prompt_template = PromptTemplate.from_template("""
分析此內部文件的合規風險，並使用即時網路資料進行驗證：

PDF 內容：
{pdf}

生成包含 PDF 引用和爬蟲法規洞察的合規報告。
""")

prompt = prompt_template.format(pdf=internal_doc)

# ------------------------------
# 使用 Langfuse 追蹤執行代理
# ------------------------------
if __name__ == "__main__":
    print("執行具有 Langfuse 追蹤功能的合規 AI 代理...")
    
    for step in agent.stream(
        {"messages": [{"role": "user", "content": prompt}]},
        stream_mode="values",
        config={"callbacks": [langfuse_handler]}
    ):
        step["messages"][-1].pretty_print()

步驟 #11：執行代理

  1. 將一份合規相關的 PDF（例如 data-processing-workflow.pdf）放置於 ./input 資料夾中。
  2. 執行代理：
python agent.py

代理將會：

  • 分析 PDF 以識別法規風險（例如「資料保留」）。
  • 使用 IPFLY 的 SERP 工具搜尋更新後的規則（例如「GDPR data retention」）。
  • 使用 IPFLY 的網路爬蟲工具爬蟲頂級政府網站（例如 europa.eu）。
  • 生成包含引用的合規報告。
  • Langfuse 將自動追蹤每一步——從 IPFLY 代理呼叫到 LLM 輸出。

步驟 #12：在 Langfuse 中檢查代理追蹤

  1. 登入您的 Langfuse 儀表板。
  2. 前往「追蹤」標籤頁——您將看到代理執行的全新追蹤記錄。
  3. 點擊追蹤記錄以探索：
    • 工具呼叫：檢視 IPFLY SERP/爬蟲請求，包含代理使用情況和回應資料。
    • LLM 互動：檢查提示、輸出和延遲。
    • 指標：追蹤爬蟲成功率、LLM 成本和總執行時間。

Langfuse 的關鍵洞察：

  • 驗證 IPFLY 代理效能（例如：爬蟲政府網站的成功率達 100%）。
  • 識別瓶頸（例如：SERP 搜尋延遲——將 IPFLY 代理類型調整為資料中心以提升速度）。
  • 稽核合規性（例如：確認代理僅使用 IPFLY 爬蟲的政府資料）。

強化代理的後續步驟

  1. 提示管理：使用 Langfuse 的提示庫對合規提示進行版本控制。
  2. 自訂 Langfuse 儀表板：追蹤 IPFLY 代理成功率、LLM 成本和合規報告品質。
  3. IPFLY 代理最佳化：對重複性爬蟲（例如每月 GDPR 更新）使用靜態住宅代理以提升一致性。
  4. 報告匯出：新增邏輯將合規報告儲存為 PDF 以供稽核。
  5. 多區域支援：使用 IPFLY 的區域 IP 爬蟲多國法規（例如：美國 CCPA、加拿大 PIPEDA）。

結論

將 Langfuse 與由 IPFLY 代理驅動的 LangChain AI 代理整合，可提供企業級的可觀測性與可靠性——這對合規使用案例至關重要。Langfuse 提供了追蹤代理每個動作的透明度，而 IPFLY 則確保不受限制地存取高品質的法規資料。

這些工具共同解決了企業 AI 代理的最大挑戰：

  • 資料存取：IPFLY 的 9000 多萬個全球代理可繞過封鎖和地理限制。
  • 可觀測性：Langfuse 追蹤每一步以供稽核和最佳化。
  • 合規性：資料來源與代理邏輯的不可變記錄。

無論您正在建構合規代理、市場研究工具或客戶支援機器人，IPFLY + Langfuse + LangChain 的組合都能創造出強大、透明且可擴展的技術堆疊。

準備好建構您自己的可觀測 AI 代理了嗎？從 IPFLY 的免費試用、Langfuse 的免費方案以及本指南的程式碼開始——為企業 AI 解鎖網路資料的全部潛力。

正文完
IPFLY 跨境電商
发表至： 跨境乾貨
近一天内
 0
IPFLY
IPFLY
高質量代理的領先提供商
用户数
2
文章数
1957
评论数
0
阅读量
910858
網頁測試最佳瀏覽器：科學剖析數位卓越的工具與技術
簡單易學的快速批量下載TikTok視頻的方法
Twitter推特X網頁版攻略：註冊、登錄及常用功能
2025海外推廣平臺特點一覽：TikTok/facebook/Instagram/YouTube