Desenvolvimento de Agentes

"""
Agente básico usando OpenAI Function Calling (tool use nativo).
Implementa o ciclo ReAct: Reason → Act → Observe → Repeat.
"""
import json
import os
import math
from datetime import datetime
from typing import Any, Callable
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

# ── Definição das Ferramentas ─────────────────────────────────────────
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Busca informações na web sobre um tópico específico. Use quando precisar de dados atualizados.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Termo de busca"},
                    "max_results": {"type": "integer", "description": "Número máximo de resultados", "default": 5},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Executa cálculos matemáticos. Aceita expressões Python seguras.",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Expressão matemática (ex: '(100 * 1.05) ** 12')"},
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_date",
            "description": "Retorna a data e hora atual. Use quando precisar saber a data de hoje.",
            "parameters": {"type": "object", "properties": {}},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "save_to_file",
            "description": "Salva conteúdo em um arquivo de texto.",
            "parameters": {
                "type": "object",
                "properties": {
                    "filename": {"type": "string", "description": "Nome do arquivo"},
                    "content": {"type": "string", "description": "Conteúdo a salvar"},
                },
                "required": ["filename", "content"],
            },
        },
    },
]

# ── Implementação das Ferramentas ─────────────────────────────────────
def search_web(query: str, max_results: int = 5) -> str:
    """Simulação de busca web (substitua por Tavily, Brave, etc.)"""
    # Em produção: use tavily-python ou similar
    return f"[BUSCA SIMULADA para '{query}']: Resultados aqui incluiriam dados reais da web."

def calculate(expression: str) -> str:
    """Executa cálculos de forma segura."""
    try:
        # Namespace seguro (apenas math e operações básicas)
        safe_globals = {"__builtins__": {}, "math": math, "abs": abs, "round": round}
        result = eval(expression, safe_globals)
        return f"{result:.4f}" if isinstance(result, float) else str(result)
    except Exception as e:
        return f"Erro no cálculo: {e}"

def get_current_date() -> str:
    return datetime.now().strftime("%d/%m/%Y %H:%M:%S")

def save_to_file(filename: str, content: str) -> str:
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"Arquivo '{filename}' salvo com sucesso ({len(content)} caracteres)."

TOOL_FUNCTIONS: dict[str, Callable] = {
    "search_web": search_web,
    "calculate": calculate,
    "get_current_date": get_current_date,
    "save_to_file": save_to_file,
}

# ── Loop do Agente ────────────────────────────────────────────────────
def run_agent(objective: str, max_iterations: int = 10) -> str:
    """
    Executa o loop ReAct até o agente responder diretamente (sem tool call).

    Args:
        objective: Objetivo em linguagem natural
        max_iterations: Limite de ciclos para evitar loops infinitos

    Returns:
        Resposta final do agente
    """
    messages = [
        {
            "role": "system",
            "content": (
                "Você é um agente inteligente com acesso a ferramentas. "
                "Use as ferramentas quando necessário para completar o objetivo. "
                "Pense passo a passo antes de cada ação. "
                "Quando tiver a resposta completa, responda diretamente sem chamar ferramentas."
            ),
        },
        {"role": "user", "content": objective},
    ]

    for iteration in range(max_iterations):
        print(f"\n🔄 Iteração {iteration + 1}")

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )

        message = response.choices[0].message
        messages.append(message)

        # Agente quer chamar ferramentas
        if message.tool_calls:
            for tool_call in message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                print(f"  🛠️  Chamando: {tool_name}({tool_args})")

                # Executar ferramenta
                fn = TOOL_FUNCTIONS.get(tool_name)
                if fn:
                    result = fn(**tool_args)
                else:
                    result = f"Ferramenta '{tool_name}' não encontrada."

                print(f"  📤 Resultado: {result[:100]}...")

                # Adicionar resultado ao histórico
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })

        # Agente terminou (resposta direta sem tool call)
        elif message.content:
            print(f"\n✅ Agente concluiu após {iteration + 1} iterações.")
            return message.content

    return "Limite máximo de iterações atingido sem conclusão."


# ── Exemplos ──────────────────────────────────────────────────────────
if __name__ == "__main__":
    result = run_agent(
        "Qual é a data de hoje? Calcule quantos dias se passaram desde 01/01/2025. "
        "Salve o resultado em 'relatorio_dias.txt'."
    )
    print(f"\n📋 Resultado Final:\n{result}")

"""
Agente stateful com LangGraph que implementa o ciclo:
Gerar → Validar → Refinar (até aprovação ou limite de iterações)
"""
from typing import Annotated, TypedDict, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
critic_llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── Estado do Grafo ───────────────────────────────────────────────────
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # Histórico de mensagens
    objective: str                            # Objetivo original
    current_draft: Optional[str]              # Rascunho atual
    critique: Optional[str]                  # Crítica do validador
    iteration: int                            # Contador de iterações
    approved: bool                            # Se o output foi aprovado
    max_iterations: int                       # Limite de ciclos

# ── Nós do Grafo ──────────────────────────────────────────────────────
def generate_node(state: AgentState) -> dict:
    """Gera ou refina o draft baseado na crítica anterior."""
    prompt = f"Objetivo: {state['objective']}\n"

    if state.get("critique"):
        prompt += f"\nRascunho anterior:\n{state['current_draft']}"
        prompt += f"\n\nCrítica recebida:\n{state['critique']}"
        prompt += "\n\nRefinamento necessário: melhore o rascunho endereçando todos os pontos da crítica."
    else:
        prompt += "\nCrie um primeiro rascunho de alta qualidade."

    response = llm.invoke([HumanMessage(content=prompt)])
    return {
        "current_draft": response.content,
        "messages": [AIMessage(content=f"[Draft {state['iteration'] + 1}] {response.content[:100]}...")],
        "iteration": state["iteration"] + 1,
    }

def validate_node(state: AgentState) -> dict:
    """Valida o draft e decide se aprova ou solicita refinamento."""
    critique_prompt = f"""Avalie criticamente o seguinte output para o objetivo dado.

Objetivo: {state['objective']}

Output para avaliação:
{state['current_draft']}

Avalie em escala 1-10 e indique se deve ser APROVADO ou REFINADO.
Se REFINADO, liste os pontos específicos a melhorar.
Se APROVADO (score >= 8), confirme "APROVADO" na primeira linha.

Formato:
APROVADO/REFINADO
Score: X/10
[Feedback detalhado]"""

    response = critic_llm.invoke([HumanMessage(content=critique_prompt)])
    critique_text = response.content
    is_approved = critique_text.upper().startswith("APROVADO")

    return {
        "critique": critique_text,
        "approved": is_approved,
        "messages": [AIMessage(content=f"[Critic] {'✅ APROVADO' if is_approved else '🔄 REFINAMENTO NECESSÁRIO'}")],
    }

def should_continue(state: AgentState) -> str:
    """Decide se continua refinando ou finaliza."""
    if state["approved"]:
        return "end"
    if state["iteration"] >= state["max_iterations"]:
        return "end"  # Limite atingido — usa melhor draft disponível
    return "generate"  # Continua refinando

# ── Construção do Grafo ───────────────────────────────────────────────
def build_refinement_agent():
    graph = StateGraph(AgentState)

    # Adicionar nós
    graph.add_node("generate", generate_node)
    graph.add_node("validate", validate_node)

    # Adicionar arestas
    graph.add_edge(START, "generate")
    graph.add_edge("generate", "validate")
    graph.add_conditional_edges(
        "validate",
        should_continue,
        {"generate": "generate", "end": END},
    )

    # Persistência de estado (permite pausar e retomar)
    checkpointer = MemorySaver()
    return graph.compile(checkpointer=checkpointer)

# ── Execução ──────────────────────────────────────────────────────────
if __name__ == "__main__":
    agent = build_refinement_agent()

    initial_state = AgentState(
        messages=[],
        objective="Escreva um pitch de 3 parágrafos para um produto SaaS de gestão financeira para MEIs, focado em simplicidade e preço acessível.",
        current_draft=None,
        critique=None,
        iteration=0,
        approved=False,
        max_iterations=3,
    )

    config = {"configurable": {"thread_id": "pitch-session-1"}}
    final_state = agent.invoke(initial_state, config=config)

    print(f"\n📝 RESULTADO FINAL (Aprovado: {final_state['approved']})")
    print(f"Iterações: {final_state['iteration']}")
    print(f"\n{final_state['current_draft']}")

"""
Crew de análise de mercado com 3 agentes especializados:
- Pesquisador: coleta informações
- Analista: analisa e interpreta
- Redator: produz o relatório final
"""
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# Ferramentas
search_tool = SerperDevTool()  # Precisa de SERPER_API_KEY

# ── Agentes ───────────────────────────────────────────────────────────
researcher = Agent(
    role="Pesquisador de Mercado Senior",
    goal="Coletar dados precisos e atualizados sobre o mercado-alvo",
    backstory=(
        "Você tem 10 anos de experiência em pesquisa de mercado. "
        "É meticuloso, cita fontes e distingue fatos de especulações. "
        "Você pesquisa exaustivamente antes de concluir qualquer análise."
    ),
    tools=[search_tool],
    llm="gpt-4o",
    verbose=True,
    memory=True,
    max_iter=10,
)

analyst = Agent(
    role="Analista de Negócios Estratégico",
    goal="Transformar dados brutos em insights acionáveis e estratégicos",
    backstory=(
        "MBA pela Wharton, 15 anos analisando mercados emergentes. "
        "Especializado em identificar oportunidades ocultas e riscos subestimados. "
        "Usa frameworks como Porter 5 Forces, SWOT e Blue Ocean Strategy."
    ),
    llm="gpt-4o",
    verbose=True,
    memory=True,
)

writer = Agent(
    role="Redator de Relatórios Executivos",
    goal="Produzir relatórios claros, concisos e convincentes para C-level",
    backstory=(
        "Ex-consultor McKinsey, especializado em comunicação executiva. "
        "Transforma análises complexas em narrativas simples e impactantes. "
        "Sabe o que CEOs e investidores querem ler."
    ),
    llm="gpt-4o",
    verbose=True,
)

# ── Tasks ─────────────────────────────────────────────────────────────
research_task = Task(
    description=(
        "Pesquise o mercado de {target_market} no Brasil para {year}. "
        "Colete: tamanho do mercado, taxa de crescimento, principais players, "
        "tendências emergentes e barreiras de entrada. "
        "Cite pelo menos 5 fontes confiáveis."
    ),
    expected_output=(
        "Relatório de pesquisa com dados quantitativos, lista de players "
        "e tendências com fontes citadas. Mínimo 500 palavras."
    ),
    agent=researcher,
)

analysis_task = Task(
    description=(
        "Com base na pesquisa fornecida, realize uma análise estratégica de {target_market}. "
        "Identifique: oportunidades de mercado, ameaças principais, "
        "gaps não endereçados pelos players atuais e recomendações estratégicas."
    ),
    expected_output=(
        "Análise SWOT, Porter 5 Forces resumido, 3 oportunidades rankeadas "
        "por potencial e 3 recomendações estratégicas concretas."
    ),
    agent=analyst,
    context=[research_task],  # Depende do resultado da pesquisa
)

writing_task = Task(
    description=(
        "Escreva um relatório executivo de 2-3 páginas sobre {target_market} "
        "para apresentar ao board. Use os dados da pesquisa e insights da análise. "
        "Inclua: sumário executivo, highlights de mercado, análise estratégica e próximos passos."
    ),
    expected_output=(
        "Relatório executivo em português, formatado com headers, "
        "bullets para os pontos principais e conclusão acionável."
    ),
    agent=writer,
    context=[research_task, analysis_task],
    output_file="market_report.md",
)

# ── Crew ──────────────────────────────────────────────────────────────
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # Ou Process.hierarchical para orquestração
    verbose=True,
    memory=True,
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"},
    },
)

if __name__ == "__main__":
    result = crew.kickoff(inputs={
        "target_market": "gestão financeira para pequenas empresas",
        "year": "2025",
    })
    print(result.raw)

"""
Agente Google ADK com:
- Subagentes especializados
- Ferramentas customizadas
- Observabilidade via Cloud Trace
- Validação e repetição automática
"""
import asyncio
from typing import AsyncGenerator
from google.adk.agents import Agent, SequentialAgent, ParallelAgent
from google.adk.tools import FunctionTool, google_search
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.events import Event
from google.genai import types

# ── Ferramentas ───────────────────────────────────────────────────────
def analyze_financial_data(data: str, metric: str) -> dict:
    """
    Analisa dados financeiros e retorna métricas calculadas.

    Args:
        data: Dados financeiros em formato CSV ou JSON string
        metric: Métrica a calcular: 'growth', 'margin', 'roi'

    Returns:
        Dicionário com resultado e contexto
    """
    # Simulação de análise — em produção conectaria ao banco de dados
    results = {
        "growth": {"value": 23.5, "period": "YoY", "unit": "%"},
        "margin": {"value": 18.2, "period": "Q4 2024", "unit": "%"},
        "roi": {"value": 3.4, "period": "12 months", "unit": "x"},
    }
    return results.get(metric, {"error": f"Metric '{metric}' not supported"})

def validate_output(content: str, criteria: list[str]) -> dict:
    """
    Valida se um output atende a critérios específicos.

    Returns:
        {'approved': bool, 'score': float, 'feedback': str}
    """
    # Em produção: use LLM para validação semântica
    word_count = len(content.split())
    criteria_met = all(
        criterion.lower() in content.lower()
        for criterion in criteria
    )
    score = min(10.0, word_count / 50)  # Score simplificado

    return {
        "approved": criteria_met and score >= 7,
        "score": score,
        "feedback": "Output aprovado" if criteria_met else f"Critérios não atendidos: {criteria}",
    }

# ── Ferramentas ADK ───────────────────────────────────────────────────
financial_tool = FunctionTool(func=analyze_financial_data)
validation_tool = FunctionTool(func=validate_output)

# ── Subagentes Especializados ─────────────────────────────────────────
research_subagent = Agent(
    name="research_agent",
    model="gemini-2.0-flash",
    description="Especialista em pesquisa e coleta de dados",
    instruction=(
        "Você é um pesquisador meticuloso. Colete informações precisas usando "
        "as ferramentas disponíveis. Sempre cite fontes e indique nível de confiança."
    ),
    tools=[google_search],
)

analysis_subagent = Agent(
    name="analysis_agent",
    model="gemini-2.0-flash",
    description="Especialista em análise financeira e estratégica",
    instruction=(
        "Você é um analista financeiro senior. Use os dados fornecidos para "
        "calcular métricas relevantes e gerar insights estratégicos acionáveis."
    ),
    tools=[financial_tool],
)

validation_subagent = Agent(
    name="validation_agent",
    model="gemini-2.0-flash",
    description="Responsável pela qualidade e validação de outputs",
    instruction=(
        "Você é um quality assurance agent. Valide se outputs atendem aos "
        "critérios especificados. Seja rigoroso — rejeite outputs incompletos."
    ),
    tools=[validation_tool],
)

# ── Orquestrador Principal ────────────────────────────────────────────
orchestrator = Agent(
    name="main_orchestrator",
    model="gemini-2.0-flash",
    description="Orquestrador principal que coordena os subagentes",
    instruction=(
        "Você é o orquestrador de um sistema de análise de negócios. "
        "Coordene os subagentes para produzir análises de alta qualidade:\n"
        "1. Use research_agent para coleta de dados\n"
        "2. Use analysis_agent para análise e métricas\n"
        "3. Use validation_agent para garantir qualidade\n"
        "4. Se a validação falhar, itere até aprovação ou 3 tentativas\n"
        "Produza um relatório final estruturado."
    ),
    sub_agents=[research_subagent, analysis_subagent, validation_subagent],
)

# ── Pipeline Sequencial Alternativo ──────────────────────────────────
sequential_pipeline = SequentialAgent(
    name="sequential_analysis",
    description="Pipeline sequencial: pesquisa → análise → validação",
    sub_agents=[research_subagent, analysis_subagent, validation_subagent],
)

# ── Pipeline Paralelo ─────────────────────────────────────────────────
parallel_pipeline = ParallelAgent(
    name="parallel_research",
    description="Múltiplas pesquisas em paralelo para maior velocidade",
    sub_agents=[research_subagent, research_subagent],  # 2 instâncias
)

# ── Runner com Observabilidade ────────────────────────────────────────
async def run_with_observability(query: str) -> None:
    """Executa o agente com logging de observabilidade."""
    session_service = InMemorySessionService()

    session = await session_service.create_session(
        app_name="business-analysis-agent",
        user_id="user-001",
    )

    runner = Runner(
        agent=orchestrator,
        app_name="business-analysis-agent",
        session_service=session_service,
    )

    print(f"🚀 Iniciando análise: {query[:100]}...")
    print(f"📍 Session ID: {session.id}")

    user_message = types.Content(
        role="user",
        parts=[types.Part.from_text(query)],
    )

    async for event in runner.run_async(
        user_id="user-001",
        session_id=session.id,
        new_message=user_message,
    ):
        # Log de cada evento do agente
        if event.is_final_response():
            print("\n✅ Resposta Final:")
            print(event.content.parts[0].text)
        elif hasattr(event, 'author') and event.author:
            # Log de ações intermediárias
            print(f"  [{event.author}] Processando...")

    # Recuperar histórico completo da sessão
    final_session = await session_service.get_session(
        app_name="business-analysis-agent",
        user_id="user-001",
        session_id=session.id,
    )

    print(f"\n📊 Estatísticas da sessão:")
    print(f"  Mensagens: {len(final_session.events)}")
    # Em produção: envie métricas para Cloud Monitoring


# ── Entry Point ───────────────────────────────────────────────────────
if __name__ == "__main__":
    asyncio.run(run_with_observability(
        "Analise o mercado de healthtech no Brasil em 2025. "
        "Inclua tamanho do mercado, crescimento YoY, principais players "
        "e 3 oportunidades de investimento. Valide se o relatório tem "
        "pelo menos 400 palavras e cobre todos os pontos solicitados."
    ))