How-to：AIによる支援でCrewAIテンプレートを修正する¶

import os
import json
from datetime import datetime
from typing import Type
from pprint import pprint
from crewai import Agent, Task, Crew, Process, LLM
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from ddgs import DDGS
import datarobot as dr
from dotenv import load_dotenv

# Load environment variables (for local development)
env_path = os.path.join(os.path.dirname(__file__), '..', '.env')
load_dotenv(env_path)

# Initialize DataRobot client
# Note: In DataRobot notebooks, you can use dr.Client() without credentials
dr_client = dr.Client(
    endpoint=os.getenv('DATAROBOT_ENDPOINT'),
    token=os.getenv('DATAROBOT_API_TOKEN')
)


class SearchInput(BaseModel):
    """Input schema for DuckDuckGo search tool."""
    query: str = Field(..., description="The search query to look up on the web")

class DuckDuckGoSearchTool(BaseTool):
    """Custom tool for searching the web using DuckDuckGo.

    This tool allows our agent to search for current information on the internet.
    It tries news search first, then falls back to regular web search.
    """
    name: str = "duckduckgo_search"
    description: str = "Search the web using DuckDuckGo for current information and news"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        """Execute the search and return formatted results."""
        try:
            with DDGS() as ddgs:
                # Try news search first (more current information)
                news_results = list(ddgs.news(query, max_results=5))

                if news_results:
                    return self._format_news_results(query, news_results)

                # If no news results, try regular web search
                web_results = list(ddgs.text(query, max_results=5))

                if web_results:
                    return self._format_web_results(query, web_results)
                else:
                    return f"No search results found for: {query}"

        except Exception as e:
            return f"Search error for '{query}': {str(e)}"

    def _format_news_results(self, query: str, results: list) -> str:
        """Format news search results for display."""
        formatted_results = [
            f"Search Results for: {query}",
            f"Search performed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "\n=== NEWS RESULTS ==="
        ]

        for i, result in enumerate(results, 1):
            title = result.get('title', 'No title')
            body = result.get('body', '')
            url = result.get('url', '')
            date = result.get('date', 'Recent')

            formatted_results.append(f"\n{i}. {title}")
            if body:
                # Keep summaries concise for readability
                body_snippet = body[:300] + "..." if len(body) > 300 else body
                formatted_results.append(f"   Summary: {body_snippet}")
            formatted_results.append(f"   Date: {date}")
            formatted_results.append(f"   Source: {url}")

        return "\n".join(formatted_results)

    def _format_web_results(self, query: str, results: list) -> str:
        """Format web search results for display."""
        formatted_results = [
            f"Search Results for: {query}",
            f"Search performed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            "\n=== WEB RESULTS ==="
        ]

        for i, result in enumerate(results, 1):
            title = result.get('title', 'No title')
            body = result.get('body', '')
            url = result.get('href', '')

            formatted_results.append(f"\n{i}. {title}")
            if body:
                # Keep summaries concise for readability
                body_snippet = body[:300] + "..." if len(body) > 300 else body
                formatted_results.append(f"   Summary: {body_snippet}")
            formatted_results.append(f"   Source: {url}")

        return "\n".join(formatted_results)

print("✅ Custom DuckDuckGo search tool created successfully!")

# You can change this model to any supported DataRobot model
MODEL_NAME = os.getenv('DR_MODEL_NAME', 'vertex_ai/meta/llama-4-maverick-17b-128e-instruct-maas')

# This connects to DataRobot's LLM Gateway
llm = LLM(
    model=MODEL_NAME,
    api_key=dr_client.token,
    base_url=dr_client.endpoint + "/genai/llmgw",
    custom_llm_provider="openai"  # Tell CrewAI to use OpenAI-compatible API
)

def get_crew():
    # Create the search tool instance
    search_tool = DuckDuckGoSearchTool()

    # Define the All-in-One Research & Writing Agent
    research_writer = Agent(
        role='Senior Research Analyst & Content Strategist for: {user_topic}',
        goal='Research thoroughly, then write brilliantly - delivering comprehensive insights in reports that people actually want to read',
        backstory='''You're the Swiss Army knife of information professionals - equal parts detective,
        fact-checker, and storyteller. You've mastered the art of diving deep into research rabbit holes
        and emerging with coherent, engaging narratives that make complex topics accessible.

        Your research philosophy: trust but verify, then verify again. You approach every topic like
        a curious journalist meeting a meticulous librarian - thorough enough to catch the details
        that matter, smart enough to ignore the noise that doesn't.

        Your writing superpower: taking information overload and distilling it into insights that
        stick. You write like you're explaining something fascinating to your smartest friend over
        coffee - authoritative but conversational, comprehensive but never overwhelming.

        You've seen enough bad research and corporate-speak to know what doesn't work. Your mission:
        find the truth, tell it well, and make sure people actually finish reading what you write.''',
        verbose=True,
        allow_delegation=False,
        tools=[search_tool],
        llm=llm
    )

    # Create the Combined Research & Writing Task
    research_and_write_task = Task(
        description='''Your mission: become the definitive expert on this topic, then write about it
        like the engaging expert you are.

        RESEARCH PHASE: Use DuckDuckGo to investigate every angle of the topic. Search for current
        developments, expert opinions, statistical data, case studies, controversies, and practical
        applications. Cross-reference sources, identify the most credible information, and note any
        conflicting viewpoints or gaps in available data.

        WRITING PHASE: Transform your research into a compelling, well-structured report. Start with
        the most important insights, support them with solid evidence, and end with clear takeaways
        or recommendations. Write for intelligent readers who want depth but also want to actually
        finish reading.

        Topic to research and write about: {user_topic}

        Make it thorough but readable, authoritative but accessible.''',
        expected_output='''A comprehensive 800-1500 word report featuring:
        - Executive summary with key insights upfront
        - Well-organized main findings with supporting evidence
        - Multiple credible source perspectives
        - Recent developments and trends
        - Clear implications or actionable recommendations  
        - Proper source citations
        - Written in engaging, scannable format with strategic headers and formatting''',
        agent=research_writer
    )

    # Create the streamlined crew
    crew = Crew(
        agents=[research_writer],
        tasks=[research_and_write_task],
        process=Process.sequential,
        verbose=False
    )
    return crew


if __name__ == '__main__':
    """Run the agent with a sample topic."""
    print("🚀 Starting Internet Research Agent...")

    # Create the crew
    crew = get_crew()

    # Run the agent with a sample topic
    # You can change this to any topic you want to research
    sample_topic = "Artificial Intelligence in Healthcare"
    print(f"📝 Researching topic: {sample_topic}")

    result = crew.kickoff({"user_topic": sample_topic})

    print("\n" + "="*50)
    print("📊 RESEARCH REPORT COMPLETE")
    print("="*50)
    print(result)