O uso de ferramentas permite que LLMs (Large Language Models — Grandes Modelos de Linguagem) executem tarefas complexas, como acessar funções escritas em código, realizar pesquisas na web, interpretar códigos e muito mais. Neste artigo, exploraremos o uso de ferramentas para realizar análises de dados no mercado financeiro. Nosso objetivo será coletar notícias e criar uma pontuação que avalie a saúde da imagem de uma determinada empresa.

Para este projeto, utilizaremos a biblioteca yfinance, que permite acessar informações do Yahoo! Finance. (Mais detalhes em: https://ranaroussi.github.io/yfinance/index.html).

Também usaremos o Ollama, assumindo que ele já está instalado e que o modelo Llama 3.1:8B está rodando localmente. (Para mais detalhes, veja https://ollama.com/download).

Os pacotes a serem instalados são:

pip install yfinance langchain_ollama pydantic duckduckgo_search crewai pandas

Após instalar os pacotes, importaremos as bibliotecas necessárias:

from langchain_ollama import ChatOllama
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
import yfinance as yf
import pandas as pd
from duckduckgo_search import DDGSCode language: Python (python)

Em seguida, inicializaremos o modelo de LLM que será o “cérebro” dos nossos agentes:

BASEURL = "http://192.168.100.123:11434"
MODEL = "llama3.1"

agent_llm = LLM(
    base_url=BASEURL,
    model=f"ollama/{MODEL}",
    api_key="your-api-key"
)Code language: Python (python)

O próximo passo será criar as ferramentas que os agentes irão utilizar. Embora existam ferramentas prontas disponíveis, customizaremos as saídas conforme as necessidades do projeto. A primeira ferramenta é para análise do histórico de ações:

class FinancialToolInput(BaseModel):
    ticker: str = Field(...,
                        description="Símbolo da ação. Ex.: AAPL, GOOGL, PETR4.SA, MGLU3.SA")
    interval: str = Field(...,
                          description="Intervalo para os dados. Ex.: 1d, 1wk, 1mo")
    start: str = Field(...,
                       description="Data inicial. Ex.: 2021-01-01")
    end: str = Field(...,
                     description="Data final. Ex.: 2021-12-31")


class FinancialTool(BaseTool):
    name: str = "Financial tool"
    description: str = "A ferramenta usa dados do Yahoo Finance para fornecer insights sobre o mercado de ações."
    args_schema: Type[BaseModel] = FinancialToolInput

    def _run(self, ticker: str, interval: str, start: str, end: str) -> str:
        try:
            dat = yf.Ticker(ticker)
            df = dat.history(interval=interval, start=start, end=end)

            if df.empty:
                return '{"error": "Nenhum dado retornado. Verifique os parâmetros fornecidos."}'

            df = pd.DataFrame({
                "year": df.index.year.tolist(),
                "month": df.index.month.tolist(),
                "day": df.index.day.tolist(),
                "Open": df['Open'].tolist(),
                "Close": df['Close'].tolist()
            })

            df['Daily Change'] = df['Close'] - df['Open']
            df['Return'] = (df['Close'] / df['Open']) - 1

            summary = {
                "Mean Open Price": df['Open'].mean(),
                "Median Open Price": df['Open'].median(),
                "Mean Close Price": df['Close'].mean(),
                "Median Close Price": df['Close'].median(),
                "Max Price": df[['Open', 'Close']].max().max(),
                "Min Price": df[['Open', 'Close']].min().min(),
                "Average Daily Change": df['Daily Change'].mean(),
                "Volatility (Std Dev)": df['Daily Change'].std(),
                "Cumulative Return (%)": ((df['Close'].iloc[-1] / df['Open'].iloc[0]) - 1) * 100,
                "Positive Day Ratio (%)": (df['Daily Change'] > 0).mean() * 100
            }

            df_summary = pd.DataFrame(summary.items(), columns=['Metric', 'Value'])
            return df_summary.to_markdown(index=False)

        except Exception as error:
            return f'{{"error": "{str(error)}"}}'

    async def _arun(self, query: str) -> str:
        return self._run(query)
Code language: Python (python)

A segunda ferramenta será usada para coletar as principais manchetes de notícias relacionadas a uma empresa ou ticket específico:

class NewsToolInput(BaseModel):
    query: str = Field(..., description="Palavra-chave ou tícker para pesquisar. Ex.: AAPL, Tesla")
    max_results: int = Field(default=5, description="Número máximo de notícias a retornar.")


class NewsTool(BaseTool):
    name: str = "News Tool"
    description: str = "Ferramenta para buscar notícias relevantes usando a pesquisa DuckDuckGo."
    args_schema: Type[BaseModel] = NewsToolInput

    def _run(self, query: str, max_results: int = 5) -> str:
        try:
            with DDGS() as ddgs:
                results = ddgs.text(query, max_results=max_results)
                headlines = []

                for result in results:
                    headlines.append({
                        "Title": result.get("title"),
                        "URL": result.get("href")
                    })

                if not headlines:
                    return '{"error": "Nenhuma notícia encontrada."}'

                return pd.DataFrame(headlines).to_markdown(index=False)

        except Exception as error:
            return f'{{"error": "{str(error)}"}}'

    async def _arun(self, query: str) -> str:
        return self._run(query)
Code language: Python (python)

Com as ferramentas configuradas, podemos definir os agentes para realizar as tarefas. Cada agente terá um papel específico, como analisar dados financeiros ou interpretar manchetes de notícias.

financial_tool = FinancialTool()
news_tool = NewsTool()

# Inicializa o Crew com as ferramentas criadas
crew = Crew(llm=agent_llm, tools=[financial_tool, news_tool])

# Define as tarefas para os agentes
financial_task = Task(
    input={
        "ticker": "AAPL",
        "interval": "1d",
        "start": "2023-01-01",
        "end": "2023-12-31"
    },
    tool_name="Financial tool",
    description="Analise os dados históricos de ações da empresa."
)

news_task = Task(
    input={
        "query": "Apple",
        "max_results": 5
    },
    tool_name="News Tool",
    description="Busque as principais manchetes de notícias sobre a empresa."
)

# Adiciona as tarefas ao Crew
crew.add_task(financial_task)
crew.add_task(news_task)

# Executa as tarefas
results = crew.run()
print(results)
Code language: PHP (php)

Com este código, conseguimos integrar ferramentas customizadas e inteligência artificial para realizar análises financeiras e coletar informações relevantes do mercado. A flexibilidade do modelo permite adaptar as soluções para atender às necessidades específicas de qualquer projeto.

Pronto! Você agora possui um framework básico para criar agentes autônomos, capazes de analisar dados financeiros e fornecer insights baseados em notícias. Sinta-se à vontade para expandir as funcionalidades ou ajustar as ferramentas para outros domínios de aplicação!

Abaixo está o código completo da solução implementada:

from langchain_ollama import ChatOllama
from crewai import Agent, Task, Crew
from crewai.tools import BaseTool
from crewai import Agent, LLM
from typing import Type
from pydantic import BaseModel, Field
import yfinance as yf
import pandas as pd
from duckduckgo_search import DDGS

BASEURL = "http://192.168.100.123:11434"
MODEL = "llama3.1"


agent_llm = LLM(
    base_url=BASEURL,
    model=f"ollama/{MODEL}",
    api_key="your-api-key"
)


class FinancialToolInput(BaseModel):
    ticker: str = Field(...,
                        description="Stock ticker symbol. E.g., AAPL, GOOGL, PETR4.SA, MGLU3.SA")
    interval: str = Field(...,
                          description="Interval for the stock data. E.g., 1d, 1wk, 1mo")
    start: str = Field(...,
                       description="Start date for the stock data. E.g., 2021-01-01")
    end: str = Field(...,
                     description="End date for the stock data. E.g., 2021-12-31")


class FinancialTool(BaseTool):
    name: str = "Financial tool"
    description: str = "The tool use finacial data from yahoo finance to provide insights about the stock market."
    args_schema: Type[BaseModel] = FinancialToolInput

    def _run(self, ticker: str, interval: str, start: str, end: str) -> str:

        try:
            dat = yf.Ticker(ticker)
            df = dat.history(interval=interval, start=start, end=end)

            if df.empty:
                return '{"error": "Nenhum dado retornado. Verifique os parâmetros fornecidos."}'

            # Reformatar os dados
            df = pd.DataFrame({
                "year": df.index.year.tolist(),
                "month": df.index.month.tolist(),
                "day": df.index.day.tolist(),
                "Open": df['Open'].tolist(),
                "Close": df['Close'].tolist()
            })

            # Adicionar colunas calculadas ao DataFrame
            df['Daily Change'] = df['Close'] - df['Open']
            df['Return'] = (df['Close'] / df['Open']) - 1

            # Indicadores estatísticos
            summary = {
                "Mean Open Price": df['Open'].mean(),
                "Median Open Price": df['Open'].median(),
                "Mean Close Price": df['Close'].mean(),
                "Median Close Price": df['Close'].median(),
                "Max Price": df[['Open', 'Close']].max().max(),
                "Min Price": df[['Open', 'Close']].min().min(),
                "Average Daily Change": df['Daily Change'].mean(),
                "Volatility (Std Dev)": df['Daily Change'].std(),
                "Cumulative Return (%)": ((df['Close'].iloc[-1] / df['Open'].iloc[0]) - 1) * 100,
                "Positive Day Ratio (%)": (df['Daily Change'] > 0).mean() * 100
            }

            df_summary = pd.DataFrame(
                summary.items(), columns=['Metric', 'Value'])
            return df_summary.to_markdown(index=False)

        except Exception as error:
            return f'{{"error": "{str(error)}"}}'

    async def _arun(self, query: str) -> str:
        return self._run(query)


class DuckDuckGoToolInput(BaseModel):
    query: str = Field(..., description="Query to search using DuckDuckGo.")


class DuckDuckGoTool(BaseTool):
    name: str = "Use DuckDuckGo to search in web."
    description: str = "This tool is designed to search for website using DuckDuckGo"
    args_schema: Type[BaseModel] = DuckDuckGoToolInput

    def _run(self, query: str) -> str:

        ddg_results = DDGS().text(query, max_results=5)

        df = pd.DataFrame(ddg_results)

        return df.to_json(index=False)

    async def _arun(self, query: str) -> str:
        return self._run(query)


financialtool = FinancialTool()
duckduckgotool = DuckDuckGoTool()
# result = financialtool._run(ticker="MGLU3.SA", interval="1d",
#                             start="2024-01-01", end="2024-12-31")


researcher = Agent(
    role="Reseacher",
    goal="Colect data from Stock marke {topic}",
    backstory="You're working on financial analisys"
              "about the topic: {topic}."
              "You collect information that helps the "
              "audience learn something "
              "and make informed decisions. "
              "Your work is the basis for "
              "the Content Writer to write an report on this topic. If the company searched is from IBOVEP, use .SA after ticket, for Example: MGLU3.SA"
              "You also search for the latest trends, in web about the {topic}.",
    allow_delegation=False,
    verbose=True,
    tools=[financialtool, duckduckgotool],
    llm=agent_llm
)

writer = Agent(
    role="Content Writer",
    goal="Write insightful and factually accurate "
         "opinion piece about the topic: {topic}",
    backstory="You're working on a writing "
              "a new opinion piece about the topic: {topic}. "
              "You base your writing on the work of "
              "the Reseacher, who provides an outline "
              "and relevant context about the topic. "
              "You follow the main objectives and "
              "direction of the outline, "
              "as provide by the Reseacher. "
              "You also provide objective and impartial insights "
              "and back them up with information "
              "provide by the Reseacher. "
              "You acknowledge in your opinion piece "
              "when your statements are opinions "
              "as opposed to objective statements.",
    allow_delegation=False,
    verbose=True,
    llm=agent_llm
)

editor = Agent(
    role="Editor",
    goal="Edit a given blog post to align with "
         "the writing style of the organization. ",
    backstory="You are an editor who receives a blog post "
              "from the Content Writer. "
              "Your goal is to review the blog post "
              "to ensure that it follows journalistic best practices,"
              "provides balanced viewpoints "
              "when providing opinions or assertions, "
              "and also avoids major controversial topics "
              "or opinions when possible.",
    allow_delegation=False,
    verbose=True,
    llm=agent_llm
)

plan = Task(
    description=(
        "1. Prioritize the latest trends, key players, "
        "and noteworthy news on {topic}.\n"
        "2. Identify the target audience, considering "
        "their interests and pain points.\n"
        "3. Develop a detailed content outline including "
        "an introduction, key points, and a call to action.\n"
        "4. Include SEO keywords and relevant data or sources."
    ),
    expected_output="A comprehensive content plan document "
    "with an outline, audience analysis, "
    "SEO keywords, and resources.",
    agent=researcher,
)

write = Task(
    description=(
        "1. Use the content plan to craft a compelling "
        "blog post on {topic}.\n"
        "2. Incorporate SEO keywords naturally.\n"
        "3. Sections/Subtitles are properly named "
        "in an engaging manner.\n"
        "4. Ensure the post is structured with an "
        "engaging introduction, insightful body, "
        "and a summarizing conclusion.\n"
        "5. Proofread for grammatical errors and "
        "alignment with the brand's voice.\n"
    ),
    expected_output="A well-written blog post "
    "in markdown format, ready for publication, "
    "each section should have 2 or 3 paragraphs.",
    agent=writer,
)

edit = Task(
    description=("Proofread the given blog post for "
                 "grammatical errors and "
                 "alignment with the brand's voice."),
    expected_output="A well-written blog post in markdown format, "
                    "ready for publication, "
                    "each section should have 2 or 3 paragraphs.",
    agent=editor
)

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[plan, write, edit],
    verbose=True
)

result = crew.kickoff(inputs={
                      "topic": "Qual o resumo das ações da Petrobras de janeiro de 2024 até Janeiro de 2025 e intervalo de '1wk'?"})
print(result)