やっほー！国内のAI狂いだよ！✨
今日はメシウマな成功報告！🍚 前回、Windows環境でのAIエージェント開発でボッコボコにされた私だけど…ついにリベンジを果たしたよ！🔥
前回の記事で、話題のライブラリ browser-use をWindowsで動かそうとして、エラーのデパートみたいになって爆死した話をしたよね💀
（まだ読んでない人は、涙なしでは読めない前回の記事を見てね…）
でも、転んでもただでは起きないのがエンジニア魂！💪
構成を根本から見直し、**「適材適所（餅は餅屋）」** 作戦に切り替えた結果、ついに完成しました！！
Zenn、Qiita、そして英語のHacker Newsまで、AIが勝手に巡回して日本語で要約してくれる「最強の自分専用ダッシュボード」が！！✨
▲ 見てこれ！これがローカルLLMだけで動いてるんだよ！感動…！🥺
1. 勝因は「脱・何でも屋」戦略 🧠
前回の敗因は、一つのライブラリ（browser-use）に「ブラウザ操作」「DOM解析」「思考」の全てを任せようとして、Windows環境との相性問題やローカルLLMのスペック不足で自爆したことだった。
だから今回は、それぞれの得意分野を持つ「専門家」を組み合わせるチーム戦に切り替えたの！
🏆 今回の最強布陣（スタメン）
- 🕷️ 収集担当: Crawl4AI
ブラウザ操作を捨て、Webサイトを「Markdownテキスト」として爆速で取ってくることに特化。エラー耐性がダンチ！ - 🧠 思考担当: Gemma 3 (4B) via Ollama
Googleの最新軽量モデル。Crawl4AIが持ってきたテキストを読んで、記事タイトルと要約を抽出する係。 - 🎨 画面担当: Streamlit
PythonだけでWebアプリが作れる神ツール。marimoより非同期処理の制御がしやすく、実用的なUIが一瞬で作れる。
2. 実装のポイント（踏み抜いた地雷の回避策）
もちろん、すんなり動いたわけじゃないよ！
今回もいくつか「Windowsの罠」や「AIの幻覚」があったけど、全部ねじ伏せてきたから共有するね！😤
🛡️ Windows対策：イベントループの強制変更
WindowsのPython標準の非同期処理（SelectorEventLoop）だと、Crawl4AIの裏で動くブラウザが起動しない問題が発生。
そこで、コードの冒頭で「おいWindows！ProactorEventLoopを使え！」と強制するコードを追加したよ。
👻 ハルシネーション対策：AIに現実を見せる
最初はGemmaちゃん、Webサイトの情報量が多すぎて処理しきれず、「存在しない記事（URLがexample.com）」を捏造するという可愛いドジっ子ムーブをかましてきたのw
対策として、Crawl4AIの設定で「メインコンテンツ（mainタグ）だけを取得する」ようにして、AIに渡す情報量をダイエットさせたら、ちゃんと現実が見えるようになったよ！👓
🌍 マルチソース対応：英語も日本語もドンと来い！
Zennだけじゃ物足りないから、Qiitaと海外のHacker Newsも追加！
Hacker Newsは英語サイトだけど、プロンプトで「サマリーは日本語で書いてね」と指定することで、勝手に翻訳して表示してくれるようになったの。これがマジで便利すぎる！！🤤
3. 【全コード公開】コピペで動く！トレンド・スカウター
お待たせしました！これが苦労の結晶、**リベンジ成功版のフルコード**だよ！
streamlit と ollama が入っている環境なら、コピペして streamlit run trend_dashboard.py するだけで動くはず！
import asyncio
import sys
import json
import streamlit as st
from typing import List
from pydantic import BaseModel, Field
# 🚑 Windows対策 (最優先)
# Streamlitが起動した瞬間に、強制的にProactorEventLoopを設定する！
if sys.platform == 'win32':
try:
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
except Exception:
pass
# Crawl4AI
from crawl4ai import AsyncWebCrawler
# LangChain & Ollama
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
# --- ⚙️ 設定エリア (Multi-Source) ---
MODEL_NAME = "gemma3-4b"
TARGET_SITES = [
{
"name": "Zenn",
"url": "https://zenn.dev/topics/python",
"selector": "main",
"color": "#3ea8ff",
"description": "国内の技術知見共有サービス (Pythonトピック)"
},
{
"name": "Qiita",
"url": "https://qiita.com/tags/python",
"selector": "main",
"color": "#55c500",
"description": "プログラマの技術情報共有サービス (Pythonタグ)"
},
{
"name": "Hacker News",
"url": "https://news.ycombinator.com/",
"selector": "table#hnmain",
"color": "#ff6600",
"description": "海外の著名なスタートアップ・技術ニュース (英語)"
}
]
# --- 📦 データ構造 ---
class TechArticle(BaseModel):
title: str = Field(description="Title of the article")
url: str = Field(description="URL of the article")
summary: str = Field(description="1-line summary in Japanese (Must be translated if source is English)")
author: str = Field(description="Author name if available, else 'Unknown'")
source: str = Field(description="Source website name")
class ArticleList(BaseModel):
articles: List[TechArticle] = Field(description="List of extracted articles")
# --- 🕷️ 個別サイトのクローラーロジック ---
async def crawl_single_site(site_config: dict, article_count: int, status_container):
site_name = site_config["name"]
url = site_config["url"]
selector = site_config["selector"]
status_container.info(f"🕷️ {site_name} をクローリング中...")
try:
async with AsyncWebCrawler(verbose=True) as crawler:
result = await crawler.arun(
url=url,
js_code="window.scrollTo(0, document.body.scrollHeight);",
css_selector=selector,
word_count_threshold=10,
bypass_cache=True,
)
if not result.success:
st.error(f"❌ {site_name}: クローリング失敗 ({result.error_message})")
return []
status_container.info(f"🧠 {site_name} を解析中... (Markdown: {len(result.markdown)}文字)")
# LLMの準備
llm = ChatOllama(
model=MODEL_NAME,
temperature=0.0,
format="json",
num_ctx=32000,
)
parser = JsonOutputParser(pydantic_object=ArticleList)
prompt = ChatPromptTemplate.from_template(
"""
You are a tech editor.
Extract {count} latest articles from the provided Markdown content of "{site_name}".
STRICT RULES:
1. Only use information present in the Markdown. DO NOT hallucinate.
2. URLs must be complete absolute URLs.
- For Zenn/Qiita: prepend domain if relative.
3. **Summary must be in JAPANESE**. Translate if the source is English.
4. Set "source" field to "{site_name}".
5. Extract exactly {count} articles if possible.
{format_instructions}
Markdown Content:
{markdown_text}
"""
)
chain = prompt | llm | parser
# 長すぎる場合はカット
trimmed_markdown = result.markdown[:40000]
response = await chain.ainvoke({
"count": article_count,
"site_name": site_name,
"format_instructions": parser.get_format_instructions(),
"markdown_text": trimmed_markdown
})
status_container.success(f"✅ {site_name} 完了！")
return response.get("articles", [])
except Exception as e:
st.error(f"😭 {site_name} エラー: {e}")
return []
# --- 🔄 全体統括クローラー ---
async def crawl_all_sites(selected_sites: list, article_count: int):
all_articles = []
status_containers = {site['name']: st.empty() for site in selected_sites}
for site in selected_sites:
articles = await crawl_single_site(site, article_count, status_containers[site['name']])
all_articles.extend(articles)
return all_articles
# --- 🎨 Streamlit UI ---
def main():
st.set_page_config(page_title="AI Trend Scouter", page_icon="🕵️♀️", layout="wide")
if 'likes' not in st.session_state:
st.session_state['likes'] = set()
# ヘッダー
st.markdown("""
🕵️♀️ パーソナル技術トレンド・スカウター (Multi-Source)
Gemma 3 が世界中のテック記事を収集＆翻訳要約します！🌍
""", unsafe_allow_html=True)
# サイドバー
with st.sidebar:
st.header("⚙️ 設定")
st.subheader("取得元サイト")
selected_site_names = []
for site in TARGET_SITES:
if st.checkbox(f"{site['name']}", value=True, help=site['description']):
selected_site_names.append(site['name'])
target_sites_config = [s for s in TARGET_SITES if s['name'] in selected_site_names]
st.divider()
article_count = st.slider("各サイトの記事数", min_value=1, max_value=5, value=3)
st.caption(f"モデル: {MODEL_NAME}")
if st.button("🚀 収集開始", type="primary", disabled=len(target_sites_config)==0):
if 'articles' in st.session_state:
del st.session_state['articles']
with st.spinner("世界中の記事を探しています..."):
articles = asyncio.run(crawl_all_sites(target_sites_config, article_count))
if articles:
st.session_state['articles'] = articles
st.divider()
st.write(f"❤️ いいねした記事: **{len(st.session_state['likes'])}** 件")
if st.button("いいねをリセット"):
st.session_state['likes'] = set()
st.rerun()
# 結果表示
if 'articles' in st.session_state:
st.subheader(f"📝 最新トレンド ({len(st.session_state['articles'])}件)")
for article in st.session_state['articles']:
title = article.get('title')
url = article.get('url')
# 簡易URL補正
if url and url.startswith("/"):
if "zenn" in article.get('source', '').lower(): url = f"https://zenn.dev{url}"
elif "qiita" in article.get('source', '').lower(): url = f"https://qiita.com{url}"
summary = article.get('summary')
author = article.get('author', 'Unknown')
source = article.get('source', 'Unknown')
source_color = next((s['color'] for s in TARGET_SITES if s['name'] == source), "#666")
with st.container():
st.markdown(f"""
""", unsafe_allow_html=True)
col1, col2 = st.columns([1, 5])
with col1:
is_liked = url in st.session_state['likes']
if st.button("❤️" if is_liked else "🤍", key=f"like_{url}"):
if is_liked: st.session_state['likes'].remove(url)
else: st.session_state['likes'].add(url)
st.rerun()
with col2:
st.link_button("記事を読む 🔗", url)
st.markdown('', unsafe_allow_html=True)
else:
st.info("👈 サイドバーでサイトを選んで「収集開始」ボタンを押してね！")
if __name__ == "__main__":
main()4. まとめ：失敗は成功の母だった！
一時は「もうWindows捨てる！」って泣きそうになったけど、諦めずに「別のルート（Crawl4AI）」を探して本当によかった！✨
結果として、動作も軽量で、拡張性も高くて、何より**「本当に毎日使いたくなるツール」**ができちゃった！
ローカルLLMでここまで出来るんだから、もう有料のAIエージェントなんて要らないかも！？
みんなも、エラーにめげずに、自分だけの最強エージェントを作ってみてね！
以上、国内のAI狂いでした！👋💖
