Difyやn8nなどのツールを用いれば、AIワークフローは簡単に実装できます。しかし、お客様データ基盤内に組込みたいという要件がある場合、どのように実装していくかを考えたいと思います。
今回はGoogle Cloudのサーバレスコンピューティングサービスである、Cloud Run FunctionsにPythonで実装してみます。
技術スタック
- LangChain: LLM（大規模言語モデル）を活用したアプリケーション開発のためのフレームワーク
- LangGraph: LangChainエコシステムの一部で、複雑なマルチステップのAIワークフローを構築するためのライブラリ
- LLM (gemini-2.5-flash)：AI翻訳・要約エンジン
- Cloud Run Functions (2CPU 4GiBメモリ)： サーバーレスのクラウドコンピューティングサービス
- Webスクレイピング
機能概要
LangGraphを使ってWebスクレイピングとAIを組み合わせた自動海外ニュース収集・翻訳システムの実装を行ってみました。
Google Geminiと組み合わせることで、URLからコンテンツを収集し、日本語で要約する自動化パイプラインを構築できます。
システム構成
このシステムは4つのノードからなるワークフローで構成されています：
- URL読み込み (node_read_urls) - ハードコードされたURLの読み込み
- Webスクレイピング (node_web_search) - URLからコンテンツを収集
- 翻訳・要約 (node_translate_content) - AI による日本語翻訳と要約
- マークダウン出力 (node_write_markdown) - 結果を出力
グラフ構造は以下の様なとてもシンプル（一直線）なフローになります。
実装
1.ライブラリの読み込み、Stateの設定
import re
import markdown
from datetime import datetime
from typing import List, Any, TypedDict, Annotated, Optional
from langchain_core.documents import Document
from langchain_community.document_loaders import RecursiveUrlLoader
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash",temperature=0.0)
Geminiを使用する際は、GOOGLE_API_KEYが必要となります。Cloud Run Functionsの環境変数として事前に設定が必要です。
2.stateの設定
LangGraphはStateというデータクラスを定義します。この中に、AIとの会話履歴やインプットデータ、出力結果をインプットしていく流れになります。
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
urls: List[str]
article_docs: List[list[Document]]
retriever: Optional[Any]
3.URLリストの読み込み
URLなどの参照データは外部ファイルとして別管理すべきですが、今回はハードコードされたURLを読み込みます。
必要に応じてGCSなどから読み込むとよいでしょう。
def node_read_urls(state: AgentState) -> AgentState:
"""
機能1: URLを読み込む。
"""
print("✨️F1")
# ここでは、ハードコードされたURLを使用します。
state["urls"] = [
"<https://dev.to/t/ai/top/week>",
"<https://hackernoon.com/c/ai>"
]
return state
4.Webスクレイピング
前ノードで設定したURLからhtmlの内容を取得します。
langchain_community.document_loaders内の、RecursiveUrlLoader関数により、再帰的にドキュメントが取得できます。
max_depthで再帰数は設定できますが、多いとサイトによっては膨大な数を調べ始めますので、1-2あたりが良いかと思います。
Documentクラス形式でデータは取得されますが、html ボディのみを取得し、不要な改行は削減するように抽出関数をBeautifulSoupeを使って設定しています。
def node_web_search(state: AgentState) -> AgentState:
"""
機能2: 与えられたURLを検索する。
"""
print("✨️F2")
state["article_docs"] = []
# ドキュメントの最大長を制限する。
max_length = 10
# ドキュメントの抽出関数
from bs4 import BeautifulSoup
def bs4_extractor(html: str) -> str:
soup = BeautifulSoup(html, "lxml")
return re.sub(r"\\n\\n+", "\\n\\n", soup.text).strip()
# ドキュメントの読み込み
tmp_article_docs = []
for url in state["urls"]:
documents = RecursiveUrlLoader(
url,
max_depth=2,
extractor=bs4_extractor,
).load()
state["article_docs"].extend(documents[:max_length])
print(f" {url} から {len(documents)} ドキュメントが読み込まれました。 ")
return state
5.記事の翻訳・要約
ここでようやくLLMを活用します。systemメッセージ内にAIの役割、実行内容を記載します。
humanメッセージに、前ノードで取得した記事内容{content}を投入します。
chain = prompt | llm | StrOutputParser() の部分がLangChainの書き方になります。 promptからllmにデータを渡し、最後に文字列変換しています。
AIに投げた回答をstateのmessage へ追加しています。（messages は add_messaagesというreducerを設定しているため、以下のような書き方ができます。）
def node_translate_content(state: AgentState) -> AgentState:
""" 機能3: 検索結果を日本語に翻訳
"""
print("✨️F3")
prompt = ChatPromptTemplate.from_messages([
("system", """
あなたはドキュメント解説の専門家です。
ステップバイステップで以下の指示に従ってください。
与えられたドキュメントの内容を要約し、重要な情報を抽出して、わかりやすく日本語で説明してください。
重要な情報は、見出しやキーワード,対象URLを含めてください。
URLは、タイトル(URL)の形式で出力してください。
URLは、トピックの後に続けてください。
例）トピック(<https://example.com>)
すべてのDocumentの内容を要約してください。
もし、内容が長い場合は、要約してください。
出力は、Markdown形式で行ってください。
"""),
("human", "次の文書を要約してください。 {content}")
])
chain = prompt | llm | StrOutputParser()
res = chain.invoke({"content":state["article_docs"]})
return {"messages": res}
6.記事の保存
AIが生成した内容をマークダウンファイルとしてローカルに保存しています。（Cloud Run Functionsに実装しているためあまり意味はないです。）
必要に応じてデータを永続化する場合は、Cloud Storageに登録するなど書き換えは必要です。
def node_write_markdown(state: AgentState) -> AgentState:
print("✨️F4")
today = datetime.now().strftime("%Y-%m-%d")
with open(f"news/webscrape_{today}.md","w", encoding="utf-8") as f:
f.write(state["messages"][-1].content)
print(f"news/webscrape_{today}.md に書き込みました。")
return state
7.グラフを作成
最後に、上記の関数をノード化し、エッジでつなぐ作業を行います。
def create_graph():
"""
グラフを作成する関数
"""
### グラフの作成
graph = StateGraph(AgentState)
### ノードの作成
graph.add_node("read_urls", node_read_urls)
graph.add_node("web_search", node_web_search)
graph.add_node("translate_content", node_translate_content)
graph.add_node("write_markdown", node_write_markdown)
### エッジの作成
graph.add_edge(START, "read_urls")
graph.add_edge("read_urls","web_search")
graph.add_edge("web_search","translate_content")
graph.add_edge("translate_content","write_markdown")
graph.add_edge("write_markdown", END)
app = graph.compile()
return app
8.実行
最後にワークフローをmain関数内で実行してみます。
結果をHTMLに変換し、ブラウザで表示してみます。
import functions_framework
@functions_framework.http
def main(request):
print("✨️main")
# グラフの作成
app = create_graph()
# AIワークフロー実行
res = app.invoke({"messages":[],"urls": []})
# 結果をHTMLで表示してみる。
md = markdown.Markdown()
return md.convert(res["messages"][-1].content)
9.ブラウザでアクセス
次のようなCSSも何も無い質素なページですが、海外の記事を翻訳、要約してくれています。興味があれば本サイトに飛んでチェックしてみるとよいかと思います。
