AI Tech Daily Agent — 完全なアーキテクチャ徹底解説＆ワークフロー分析

ai-tech-daily-agent/
├── agent.py                    # メインエージェントのエントリポイント
├── config/
│   ├── __init__.py
│   └── sources.py             # 追跡対象のリポジトリ＆企業
├── protocols/
│   ├── __init__.py
│   └── chat_proto.py          # チャットプロトコルの実装
├── services/
│   ├── __init__.py
│   ├── article_service.py     # 記事生成ロジック
│   ├── company_picker.py      # 企業選定アルゴリズム
│   ├── devto_service.py       # Dev.to API 統合
│   ├── github_service.py      # GitHub API 統合
│   ├── image_search_service.py # 画像検索ロジック
│   ├── llm_service.py         # LLM 抽象化レイヤー
│   ├── publish_service.py     # 公開オーケストレーション
│   ├── web_scraper_service.py # コンテンツスクレイピング
│   └── web_search_service.py  # 検索 API のラッパー
├── tests/
│   ├── __init__.py
│   └── test_filter.py         # ユニットテスト
├── pyproject.toml             # プロジェクトの依存関係
├── uv.lock                    # 固定された依存関係のバージョン
├── .gitignore
├── README.md
├── PROJECT_DEEP_DIVE.md       # このドキュメント
└── docs/
    └── deep-dive/             # 生成されたダイアグラム画像（PNG）
        ├── architecture.png
        ├── pipeline.png
        └── data-flow.png

def generate_article(
    company: dict,
    search_data: dict,
    scraped_content: str,
    github_repos: list[dict],
    images: dict[str, str],
) -> tuple[str, str]:

async def _run_pipeline(ctx: Context) -> str:
    result = await asyncio.to_thread(run_pipeline, dry_run)
    return result

if result:
    # LLM 生成コンテンツを使用
else:
    result = _fallback_article(...)

TRACKED_COMPANIES = [...]
TRACKED_FRAMEWORK_REPOS = [...]

# セッション開始
StartSessionContent → Welcome message

# ユーザーコマンド
TextContent("generate") → Start pipeline
TextContent("status") → Show history
TextContent("help") → Show commands

# 承認
ChatAcknowledgement → Confirmation of receipt

# エージェント登録
Agent(
    name="ai-tech-daily-agent",
    port=8000,
    seed=AGENT_SEED,
    endpoint=["http://localhost:8000/submit"],
)

# メインパイプライン
def run_pipeline(dry_run: bool = False) -> str:
    1. 履歴を確認し 企業を選択
    2. Web/検索 クエリを実行
    3. GitHub リポジトリデータを取得
    4. スクレイプ して コンテンツを読み取る
    5. LLM を使用して 記事を生成
    6. 適切な 画像を見つける
    7. 必要に応じて Dev.to に 公開する
    8. 履歴を更新

def run_pipeline(dry_runbool = False) -> str:
    # ステップ1：会社の選択
    history = get_history()
    company = select_company(history, TRACKED_COMPANIES)

    # ステップ2：データ収集
    search_data = {
        "news": search_news(...),
        "web": search_web(...),
        "github": search_github(...),
    }

    # ステップ3：コンテンツの収集
    github_repos = get_all_repos()
    scraped_content = scrape_and_read(...)

    # ステップ4：記事の生成
    articlefilename = generate_article(...)

    # ステップ5：公開
    if not dry_run:
        devto_id = publish_to_devto(...)

    return result

def select_company(historylist[dict], companies: list[dict]) -> dict:
    cutoff = (datetime.now() - timedelta(days=14)).isoformat()
    recent_slugs = {h["slug"] for h in history if h["date"] >= cutoff}

    candidates = [c for c in companies if c["slug"] not in recent_slugs]

    if not candidates:
        log.warning("No candidates available after 14-day filter")
        return companies[0]

    return random.choice(candidates)

COMPANY_TRACKING = [
    {
        "name": "OpenAI",
        "slug": "openai",
        "topics": ["llm", "generative-ai", "gpt"],
    },
    {
        "name": "Anthropic",
        "slug": "anthropic",
        "topics": ["llm", "claude", "safety"],
    },
    # ... 他の企業
]

def search_news(company: str, topics: list[str]) -> list[dict]:
    """
    企業に関する最近のニュースを検索します。
    タイトル、URL、本文、日付を含むニュース項目のリストを返します。
    """
    queries = [company] + topics
    all_news = []

    for query in queries:
        results = _call_search_api(query="news:" + query)
        all_news.extend(results)

    return _deduplicate(all_news)

def search_web(company: str) -> list[dict]:
    """
    企業情報のための一般的なWeb検索を行います。
    """
    return _call_search_api(query=company)

# 生のAPIレスポンス
{
    "title": "...",
    "url": "...",
    "snippet": "...",
    "date": "...",
}

# 内部形式に変換
{
    "title": "...",
    "url": "...",
    "body": "...",
    "date": "...",
}

# 履歴ファイルを読み込む
if os.path.exists(HISTORY_FILE):
    history = json.loads(Path(HISTORY_FILE).read_text())
else:
    history = []

# 時間的フィルタを適用する
cutoff = (datetime.now() - timedelta(days=14)).isoformat()
recent_slugs = {h["slug"] for h in history if h["date"] >= cutoff}

# 会社を選択する
candidates = [c for c in TRACKED_COMPANIES if c["slug"] not in recent_slugs]
company = random.choice(candidates)

# 異なるクエリのバリエーションで並列検索
news_queries = [
    company["name"],
    company["name"] + " news",
    company["name"] + " announcement",
    *company["topics"]
]

all_news = []
for query in news_queries:
    news = search_news(query)
    all_news.extend(news)

# 結果の重複を除去する
seen_urls = set()
unique_news = [n for n in all_news if n["url"] not in seen_urls]

# 1. 追跡対象のフレームワーク（既知のリポジトリ）
frameworks = []
for repo in TRACKED_FRAMEWORK_REPOS:
    data = fetch_github_repo(repo["owner"], repo["repo"])
    release = get_latest_release(repo["owner"], repo["repo"])
    frameworks.append({...})

返却形式: {"translated": "翻訳されたHTML"}# 2. トレンドの新規リポジトリ（ディスカバリー）
trending = []
for query in SEARCH_QUERIES:
    repos = github_search_repository(query, 
                                     sort="stars",
                                     created=">7 days ago")
    trending.extend(repos)

# 検索結果から上位URLを取得
top_urls = [item["url"] for item in search_results[:10]]

# スクレイピングして内容を読み取る
scraped_text = " "
for url in top_urls:
    try:
        html = requests.get(url, timeout=15).text
        text = extract_text_from_html(html)
        scraped_text += text
        if len(scraped_text) > 10000:  # コンテンツの上限を設ける
            break
    except Exception as e:
        log.warning(f"{url} のスクレイピングに失敗しました： {e}")

# 包括的なプロンプトを構築
system_prompt = f"""
あなたはシニアのテック・ジャーナリストです...
TODAY'S FOCUS: {company_name}
ルール：
- 記事は必ず300行以上であること
- stars、funding、users といった具体的な数値を含めること
- 2〜3個のコードスニペットを含めること
- 出典へのリンクを含めること
"""

user_prompt = f"""
会社のトピック： {topics}

=== リアルタイムニュース ===
{formatted_news}

=== Web検索結果 ===
{formatted_web}

=== GITHUB 検索 ===
{formatted_github}

=== トラッキングしているリポジトリ ===
{formatted_repos}

=== スクレイピングしたコンテンツ ===
{scraped_content[:8000]}
"""

# 記事を生成
article = call_llm(system_prompt, user_prompt, 
                   temperature=0.7, 
                   max_tokens=8000)

images = {}

# ロゴを検索
logo_url = search_images(f"{company} logo official website")
if logo_url:
    images["logo"] = logo_url

# ヒーロー画像を検索
hero_url = search_images(f"{company} technology platform")
if hero_url:
    images["hero"] = hero_url
# テック画像を検索する
banner_url = search_images(f"{company} アーキテクチャ テクノロジー")
if banner_url:
    images["banner"] = banner_url

# ローカルに保存
filename = f"{slug}-{date}.md"
article_path = Path("articles") / filename
article_path.write_text(article)

# Dev.toに公開
if not dry_run and devto_api_key:
    devto_id = create_devto_article(
        title=f"{company} — 詳細調査",
        body_markdown=article,
        tags=company["topics"] + ["ai", "technology"],
        published=True
    )
    url = f"https://dev.to/{devto_username}/{slug}"
else:
    url = f"ローカル: {article_path}"

history.append({
    "name": company["name"],
    "slug": company["slug"],
    "date": datetime.now().isoformat(),
    "article_url": url,
    "devto_id": devto_id
})

# ファイルに永続化
Path(HISTORY_FILE).write_text(json.dumps(history, indent=2))

def _headers() -> dict:
    h = {
        "Accept": "application/vnd.github+json",
        "User-Agent": "AI-Tech-Daily-Agent/1.0"
    }
    token = os.getenv("GH_TOKEN") or os.getenv("GITHUB_TOKEN")
    if token:
        h["Authorization"] = f"token {token.strip()}"
    return h

def get_framework_updates() -> list[dict]:
    results = []

    for repo_info in TRACKED_FRAMEWORK_REPOS:
        # リポジトリのメタデータを取得
        resp = requests.get(
            f"https://api.github.com/repos/{owner}/{repo}",
            headers=headers,
            timeout=10
        )
        data = resp.json()

        # 最新リリースを取得
        release_info = _get_latest_release(owner, repo, headers)

        # 詳細なレコードを構築
        results.append({
            "name": f"{owner}/{repo}",
            "label": repo_info["label"],
            "url": data["html_url"],
            "description": data["description"],
            "stars": data["stargazers_count"],
            "language": data.get("language"),
            "updated_at": data.get("pushed_at"),
            "latest_release": release_info,
            "type": "tracked"
        })

    # 最近の活動でソート
    results.sort(key=lambda x: x.get("updated_at", ""), reverse=True)
    return results

def search_trending_repos() -> list[dict]:
    one_week_ago = (datetime.utcnow() - timedelta(days=7)).strftime("%Y-%m-%d")

    queries = [
        "ai agent",
        "llm agent framework",
        "mcp server",
        "agentic ai",
        "autonomous agent",
        # ... さらにクエリ
    ]

    all_repos = []

返却形式: {"translated": "翻訳されたHTML"}for query in queries:
        resp = requests.get(
            "https://api.github.com/search/repositories",
            params={
                "q": f"{query} created:>{one_week_ago}",
                "sort": "stars",
                "order": "desc",
                "per_page": 5
            },
            headers=headers
        )

        for repo in resp.json().get("items", []):
            all_repos.append({
                "name": repo["full_name"],
                "url": repo["html_url"],
                "description": repo["description"],
                "stars": repo["stargazers_count"],
                "language": repo["language"],
                "type": "trending"
            })

    # 重複を除去し、starsでソートする
    unique = list({r["name"]: r for r in all_repos}.values())
    unique.sort(key=lambda x: x["stars"], reverse=True)
    return unique[:10]

def generate_article(
    company: dict,
    search_data: dict,
    scraped_content: str,
    github_repos: list[dict],
    images: dict[str, str],
) -> tuple[str, str]:

system = f"""あなたはシニアのテックジャーナリストであり、デベロッパーアドボケイトとして "AI & Tech Daily" のための、深掘りした毎日の記事を書いてください。

TODAY'S FOCUS: {name}

Write a COMPREHENSIVE deep-dive about {name} — 今この瞬間に起きていることをすべて網羅しながら。

返却形式: {"translated": "翻訳されたHTML"}RULES:
- 記事は300行以上のマークダウンであること
- すべての内容は、提供されたリアルタイム検索データに基づくこと — 架空の事実を作らないこと
- 具体的な数値を入れること: 星の数、調達額、ユーザー数、バージョン番号
- コードスニペットを2〜3個含めること: それらのツール/プロダクトの使い方
- 出典へのリンクを含めること: [text](url)
- 指示がある場合は画像を入れること（ロゴ、ヒーロー、テック画像）
- 論拠のある意見（オピニオン）を入れること — それが開発者にとってどういう意味かを述べる
- 各セクションには、実際の十分に具体的な内容（実質的な中身）を入れること

必須セクション（## 見出し、すべて必須）:

# {name} — ディープダイブ | {human_date}

## 会社概要
## 最新ニュース & お知らせ
## プロダクト & テクノロジーディープダイブ
## GitHub & オープンソース
## はじめに — コード例
## 市場でのポジション & 競合
## 開発者への影響
## その' 次
## 要点
## リソース & リンク
"""

user = f"""{nameについてのディープダイブ記事を{human_date}のために書いてください。

会社のトピック: {topics}
{image_instructions}

=== リアルタイムニュース（今日検索） ===
{news_text}

=== WEB検索結果 ===
{web_text}

=== GITHUB検索 ===
{github_text}

=== 追跡レポジトリデータ ===
{repo_text}

=== スクレイピング済み記事内容（上位ソースから） ===
{scraped_content[:8000]}

重要: 記事を全文で書いてください。最低300行以上。指示された通りに画像を含め、コードスニペットを含めてください。上記のデータのみを使ってください。"""

def _format_news(news: list[dict]) -> str:
    """プロンプト用にニュース検索結果を整形します。"""
    lines = []
    for n in news[:15]:  # 上位15件に制限
        lines.append(f"- [{n['title']}]({n['url']})")
        if n.get("body"):
            lines.append(f"  {n['body'][:300]}")
        if n.get("date"):
            lines.append(f"  日付: {n['date']}")
        lines.append("")
    return "
".join(lines)

def _format_github(github: list[dict]) -> str:
    """プロンプト用にGitHub検索結果を整形します。"""
    lines = []
    for g in github[:8]:  # 上位8件に制限
        lines.append(f"- [{g['title']}]({g['url']})")
        lines.append(f"  {g.get('body', '')[:200]}")
    return "
".join(lines)

def _format_tracked_repos(repos: list[dict]) -> str:
    """リリース情報付きで追跡中のリポジトリを整形します。"""
    lines = []
    for r in repos:
        release = r.get("latest_release")
        rel = f" — 最新: {release['tag']}" if release else ""
        lines.append(
            f"- {r['label']} (⭐{r['stars']}:,}){rel} — "
            f"{r['description'][:150]} [{r['url']}]"
        )
    return "
".join(lines)

def _fallback_article(company, search_data, repos, images, 
                      human_date, date_str):
    """LLMが失敗した場合の基本テンプレート記事を生成します。"""

    name = company["name"]
    topics = ", ".join(company["topics"])

    # 利用可能なデータを整形する
    news_bullets = "
".join(
        f"- **{n['title']}** — {n.get('body', '')[:200]} [source]({n['url']})"
        for n in search_data.get("news", [])[:10]
    )

返却形式: {"translated": "翻訳されたHTML"}web_bullets = "
".join(
        f"- [{w['title']}]({w['url']})"
        for w in search_data.get("web", [])[:8]
    )

    repo_bullets = "
".join(
        f"- **[{r['label']}]({r['url']})** ⭐ {r['stars']}:,}"
        for r in repos[:10]
    )

    # テンプレートを構築する
    return f"""# {name} — ディープダイブ | {human_date}
{logo_img}
> {name} への毎日のディープダイブ — {topics} をカバーします。

---

{hero_img}

## 最新ニュース & お知らせ

{news_bullets}

---

## Web リソース

{web_bullets}

---

## GitHub & オープンソース

{repo_bullets}

---

## 重要なポイント

1. {name} は AI/テクノロジーの領域で進化を続けています
2. 進捗の更新のために、オープンソースのプロジェクトを監視してください
3. 最新のお知らせは公式チャネルで確認してください

---

* {date_str} に生成：[AI Tech Daily Agent](https://github.com/gautammanak1/ai-tech-daily-agent)*
"""

def call_llm(
    system: str,
    user: str,
    temperature: float = 0.7,
    max_tokens: int = 4000,
) -> str | None:
    """
    システムメッセージとユーザーメッセージを使用して LLM API を呼び出します。
    生成されたテキスト、または失敗時は None を返します。
    """

def call_llm(
    system: str,
    user: str,
    temperature: float = 0.7,
    max_tokens: int = 4000,
) -> str | None:
    try:
        # 環境変数から API キーを取得する
        api_key = os.getenv("OPENAI_API_KEY") or os.getenv("LLM_API_KEY")

        if not api_key:
            log.warning("LLM API キーが見つかりません
")
            return None

返却形式: {"translated": "翻訳されたHTML"}# API 呼び出しを行う
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4-turbo-preview",
                "messages": [
                    {"role": "system", "content": system},
                    {"role": "user", "content": user}
                ],
                "temperature": temperature,
                "max_tokens": max_tokens
            },
            timeout=60
        )

        response.raise_for_status()
        data = response.json()

        # 生成されたコンテンツを抽出する
        return data["choices"][0]["message"]["content"]

    except requests.RequestException as e:
        log.error(f"LLM API リクエストの失敗: {e}")
        return None
    except (KeyError, IndexError) as e:
        log.error(f"LLM API レスポンスの解析に失敗: {e}")
        return None

def create_devto_article(
    title: str,
    body_markdown: str,
    tags: list[str],
    published: bool = True,
) -> str | None:
    """
    Dev.to 上に記事を作成します。
    成功時は dev.to の記事 ID、失敗時は None を返します。
    """

    api_key = os.getenv("DEVTO_API_KEY")
    if not api_key:
        log.warning("Dev.to の API キーがありません")
        return Nonetry:
        response = requests.post(
            "https://dev.to/api/articles",
            headers={
                "api-key": api_key,
                "Content-Type": "application/json"
            },
            json={
                "article": {
                    "title": title,<