AI Tech Daily Agent — Complete Architecture Deep Dive & Workflow Analysis

ai-tech-daily-agent/
├── agent.py                    # Main agent entry point
├── config/
│   ├── __init__.py
│   └── sources.py             # Tracked repositories & companies
├── protocols/
│   ├── __init__.py
│   └── chat_proto.py          # Chat protocol implementation
├── services/
│   ├── __init__.py
│   ├── article_service.py     # Article generation logic
│   ├── company_picker.py      # Company selection algorithm
│   ├── devto_service.py       # Dev.to API integration
│   ├── github_service.py      # GitHub API integration
│   ├── image_search_service.py # Image finding logic
│   ├── llm_service.py         # LLM abstraction layer
│   ├── publish_service.py     # Publishing orchestration
│   ├── web_scraper_service.py # Content scraping
│   └── web_search_service.py  # Search API wrapper
├── tests/
│   ├── __init__.py
│   └── test_filter.py         # Unit tests
├── pyproject.toml             # Project dependencies
├── uv.lock                    # Locked dependency versions
├── .gitignore
├── README.md
├── PROJECT_DEEP_DIVE.md       # This document
└── docs/
    └── deep-dive/             # Generated diagram images (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:
    # Use LLM-generated content
else:
    result = _fallback_article(...)

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

# Session start
StartSessionContent → Welcome message

# User commands
TextContent("generate") → Start pipeline
TextContent("status") → Show history
TextContent("help") → Show commands

# Acknowledgments
ChatAcknowledgement → Confirmation of receipt

# Agent registration
Agent(
    name="ai-tech-daily-agent",
    port=8000,
    seed=AGENT_SEED,
    endpoint=["http://localhost:8000/submit"],
)

# Main pipeline
def run_pipeline(dry_run: bool = False) -> str:
    1. Check history and select company
    2. Perform web/search queries
    3. Fetch GitHub repository data
    4. Scrape and read content
    5. Generate article using LLM
    6. Find appropriate images
    7. Optionally publish to Dev.to
    8. Update history

def run_pipeline(dry_run: bool = False) -> str:
    # Step 1: Company Selection
    history = get_history()
    company = select_company(history, TRACKED_COMPANIES)

    # Step 2: Data Collection
    search_data = {
        "news": search_news(...),
        "web": search_web(...),
        "github": search_github(...),
    }

    # Step 3: Content Gathering
    github_repos = get_all_repos()
    scraped_content = scrape_and_read(...)

    # Step 4: Article Generation
    article, filename = generate_article(...)

    # Step 5: Publishing
    if not dry_run:
        devto_id = publish_to_devto(...)

    return result

def select_company(history: list[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"],
    },
    # ... more companies
]

def search_news(company: str, topics: list[str]) -> list[dict]:
    """
    Search for recent news about the company.
    Returns list of news items with title, url, body, date.
    """
    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]:
    """
    General web search for company information.
    """
    return _call_search_api(query=company)

# Raw API response
{
    "title": "...",
    "url": "...",
    "snippet": "...",
    "date": "...",
}

# Transformed to internal format
{
    "title": "...",
    "url": "...",
    "body": "...",
    "date": "...",
}

# Load history file
if os.path.exists(HISTORY_FILE):
    history = json.loads(Path(HISTORY_FILE).read_text())
else:
    history = []

# Apply temporal filter
cutoff = (datetime.now() - timedelta(days=14)).isoformat()
recent_slugs = {h["slug"] for h in history if h["date"] >= cutoff}

# Select company
candidates = [c for c in TRACKED_COMPANIES if c["slug"] not in recent_slugs]
company = random.choice(candidates)

# Parallel search with different query variations
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)

# Deduplicate results
seen_urls = set()
unique_news = [n for n in all_news if n["url"] not in seen_urls]

# 1. Tracked frameworks (known repos)
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({...})

# 2. Trending new repos (discovery)
trending = []
for query in SEARCH_QUERIES:
    repos = github_search_repository(query, 
                                     sort="stars",
                                     created=">7 days ago")
    trending.extend(repos)

# Get top URLs from search results
top_urls = [item["url"] for item in search_results[:10]]

# Scrape and read content
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:  # Limit content
            break
    except Exception as e:
        log.warning(f"Failed to scrape {url}: {e}")

# Build comprehensive prompt
system_prompt = f"""
You are a senior tech journalist...
TODAY'S FOCUS: {company_name}
RULES:
- Article MUST be 300+ lines
- Include specific numbers: stars, funding, users
- Include 2-3 code snippets
- Include links to sources
"""

user_prompt = f"""
Company topics: {topics}

=== REAL-TIME NEWS ===
{formatted_news}

=== WEB SEARCH RESULTS ===
{formatted_web}

=== GITHUB SEARCH ===
{formatted_github}

=== TRACKED REPOS ===
{formatted_repos}

=== SCRAPED CONTENT ===
{scraped_content[:8000]}
"""

# Generate article
article = call_llm(system_prompt, user_prompt, 
                   temperature=0.7, 
                   max_tokens=8000)

images = {}

# Search for logo
logo_url = search_images(f"{company} logo official website")
if logo_url:
    images["logo"] = logo_url

# Search for hero image
hero_url = search_images(f"{company} technology platform")
if hero_url:
    images["hero"] = hero_url

# Search for tech images
banner_url = search_images(f"{company} architecture technology")
if banner_url:
    images["banner"] = banner_url

# Save local copy
filename = f"{slug}-{date}.md"
article_path = Path("articles") / filename
article_path.write_text(article)

# Publish to Dev.to
if not dry_run and devto_api_key:
    devto_id = create_devto_article(
        title=f"{company} — Deep Dive",
        body_markdown=article,
        tags=company["topics"] + ["ai", "technology"],
        published=True
    )
    url = f"https://dev.to/{devto_username}/{slug}"
else:
    url = f"Local: {article_path}"

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

# Persist to file
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:
        # Fetch repository metadata
        resp = requests.get(
            f"https://api.github.com/repos/{owner}/{repo}",
            headers=headers,
            timeout=10
        )
        data = resp.json()

        # Fetch latest release
        release_info = _get_latest_release(owner, repo, headers)

        # Build comprehensive record
        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"
        })

    # Sort by recent activity
    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",
        # ... more queries
    ]

    all_repos = []

    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"
            })

    # Deduplicate and sort by 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"""You are a senior tech journalist and developer advocate writing an in-depth daily article for "AI & Tech Daily".

TODAY'S FOCUS: {name}

Write a COMPREHENSIVE deep-dive about {name} — covering everything happening RIGHT NOW.

RULES:
- Article MUST be 300+ lines of markdown
- ALL content must be based on the real-time search data provided — do NOT invent facts
- Include specific numbers: star counts, funding, users, version numbers
- Include 2-3 code snippets showing how to use their tools/products
- Include links to sources: [text](url)
- Include images where provided (logo, hero, tech images)
- Be opinionated — give your take on what this means for developers
- Every section must have real, substantial content

REQUIRED SECTIONS (## headings, ALL mandatory):

# {name} — Deep Dive | {human_date}

## Company Overview
## Latest News & Announcements
## Product & Technology Deep Dive
## GitHub & Open Source
## Getting Started — Code Examples
## Market Position & Competition
## Developer Impact
## What's Next
## Key Takeaways
## Resources & Links
"""

user = f"""Write a deep-dive article about {name} for {human_date}.

Company topics: {topics}
{image_instructions}

=== REAL-TIME NEWS (searched today) ===
{news_text}

=== WEB SEARCH RESULTS ===
{web_text}

=== GITHUB SEARCH ===
{github_text}

=== TRACKED REPOS DATA ===
{repo_text}

=== SCRAPED ARTICLE CONTENT (from top sources) ===
{scraped_content[:8000]}

IMPORTANT: Write FULL article. 300+ lines minimum. Use ONLY data from above. Include images where instructed. Include code snippets."""

def _format_news(news: list[dict]) -> str:
    """Format news search results for prompt."""
    lines = []
    for n in news[:15]:  # Limit to top 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"  Date: {n['date']}")
        lines.append("")
    return "
".join(lines)

def _format_github(github: list[dict]) -> str:
    """Format GitHub search results for prompt."""
    lines = []
    for g in github[:8]:  # Limit to top 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:
    """Format tracked repositories with release info."""
    lines = []
    for r in repos:
        release = r.get("latest_release")
        rel = f" — latest: {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):
    """Generate a basic template article if LLM fails."""

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

    # Format available data
    news_bullets = "
".join(
        f"- **{n['title']}** — {n.get('body', '')[:200]} [source]({n['url']})"
        for n in search_data.get("news", [])[:10]
    )

    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]
    )

    # Build template
    return f"""# {name} — Deep Dive | {human_date}
{logo_img}
> Daily deep dive into {name} — covering {topics}.

---

{hero_img}

## Latest News & Announcements

{news_bullets}

---

## Web Resources

{web_bullets}

---

## GitHub & Open Source

{repo_bullets}

---

## Key Takeaways

1. {name} continues to evolve in the AI/tech landscape
2. Monitor their open-source projects for updates
3. Check official channels for latest announcements

---

*Generated on {date_str} by [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:
    """
    Call LLM API with system and user messages.
    Returns generated text or None on failure.
    """

def call_llm(
    system: str,
    user: str,
    temperature: float = 0.7,
    max_tokens: int = 4000,
) -> str | None:
    try:
        # Get API key from environment
        api_key = os.getenv("OPENAI_API_KEY") or os.getenv("LLM_API_KEY")

        if not api_key:
            log.warning("No LLM API key found")
            return None

        # Make API call
        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()

        # Extract generated content
        return data["choices"][0]["message"]["content"]

    except requests.RequestException as e:
        log.error(f"LLM API request failed: {e}")
        return None
    except (KeyError, IndexError) as e:
        log.error(f"LLM API response parsing failed: {e}")
        return None

def create_devto_article(
    title: str,
    body_markdown: str,
    tags: list[str],
    published: bool = True,
) -> str | None:
    """
    Create an article on Dev.to.
    Returns dev.to article ID or None on failure.
    """

    api_key = os.getenv("DEVTO_API_KEY")
    if not api_key:
        log.warning("No Dev.to API key")
        return None

    try:
        response = requests.post(
            "https://dev.to/api/articles",
            headers={
                "api-key": api_key,
                "Content-Type": "application/json"
            },
            json={
                "article": {
                    "title": title,<