Function Calling e LlamaIndex: aprenda a criar agentes que usam ferramentas

Os agentes de IA representam um dos avanços mais significativos no desenvolvimento de aplicações com Large Language Models (LLMs). Diferente de um chatbot simples, um agente é capaz de usar ferramentas, tomar decisões e executar tarefas de forma autônoma. Neste tutorial, vamos construir um agente com function calling do zero usando LlamaIndex Workflows.

O que é function calling?

Function calling é uma funcionalidade oferecida por alguns LLMs (como OpenAI, Anthropic e Ollama) que permite ao modelo chamar funções ou usar ferramentas externas durante uma conversa. Em vez de apenas gerar texto, o LLM pode:

• Identificar quando precisa de informações adicionais

• Selecionar a ferramenta apropriada para obter essas informações

• Processar os resultados e continuar a conversa

Preparando o ambiente

Antes de começar, você precisa instalar o LlamaIndex e configurar sua API key da OpenAI:

!pip install -U llama-index

import os
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

Configurando observabilidade (opcional)

Para visualizar cada etapa do workflow, você pode configurar o tracing com Llamatrace. Como workflows são async first, tudo funciona perfeitamente em notebooks Jupyter.

Importante: Se você estiver executando fora de um notebook, use asyncio.run() para iniciar o event loop:

async def main():
    # seu código async aqui
    pass

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Arquitetura do agente: entendendo as etapas

Um agente com function calling funciona através de um ciclo bem definido:

1. Receber mensagem do usuário - adicionar à memória e recuperar histórico

2. Chamar o LLM - passar ferramentas disponíveis e histórico da conversa

3. Processar tool calls - identificar se o LLM solicitou uso de ferramentas

4. Executar ferramentas - chamar as funções necessárias e processar resultados

5. Retornar resposta - quando não há mais tool calls, retornar a resposta final

Definindo os eventos do workflow

Para implementar essas etapas, precisamos definir eventos customizados que representam cada transição no nosso workflow:

from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event

class InputEvent(Event):
    input: list[ChatMessage]

class StreamEvent(Event):
    delta: str

class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]

class FunctionOutputEvent(Event):
    output: ToolOutput

Cada evento representa um ponto de transição no nosso workflow:

Implementando o workflow completo

Agora vamos criar nosso agente. O LlamaIndex usa type annotations para validar automaticamente o workflow, então as anotações de tipo são essenciais:

from typing import Any, List
from llama_index.core.llms.function_calling import FunctionCallingLLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import (
    Context,
    Workflow,
    StartEvent,
    StopEvent,
    step,
)
from llama_index.llms.openai import OpenAI

class FuncationCallingAgent(Workflow):
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.tools = tools or []
        self.llm = llm or OpenAI()
        assert self.llm.metadata.is_function_calling_model

    @step
    async def prepare_chat_history(
        self, ctx: Context, ev: StartEvent
    ) -> InputEvent:
        # limpar fontes
        await ctx.store.set("sources", [])

        # verificar se a memória está configurada
        memory = await ctx.store.get("memory", default=None)
        if not memory:
            memory = ChatMemoryBuffer.from_defaults(llm=self.llm)

        # obter input do usuário
        user_input = ev.input
        user_msg = ChatMessage(role="user", content=user_input)
        memory.put(user_msg)

        # obter histórico de chat
        chat_history = memory.get()

        # atualizar contexto
        await ctx.store.set("memory", memory)

        return InputEvent(input=chat_history)

    @step
    async def handle_llm_input(
        self, ctx: Context, ev: InputEvent
    ) -> ToolCallEvent | StopEvent:
        chat_history = ev.input

        # fazer streaming da resposta
        response_stream = await self.llm.astream_chat_with_tools(
            self.tools, chat_history=chat_history
        )
        async for response in response_stream:
            ctx.write_event_to_stream(StreamEvent(delta=response.delta or ""))

        # salvar resposta final
        memory = await ctx.store.get("memory")
        memory.put(response.message)
        await ctx.store.set("memory", memory)

        # obter tool calls
        tool_calls = self.llm.get_tool_calls_from_response(
            response, error_on_no_tool_call=False
        )

        if not tool_calls:
            sources = await ctx.store.get("sources", default=[])
            return StopEvent(
                result={"response": response, "sources": [*sources]}
            )
        else:
            return ToolCallEvent(tool_calls=tool_calls)

    @step
    async def handle_tool_calls(
        self, ctx: Context, ev: ToolCallEvent
    ) -> InputEvent:
        tool_calls = ev.tool_calls
        tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}

        tool_msgs = []
        sources = await ctx.store.get("sources", default=[])

        # chamar ferramentas com tratamento de erros
        for tool_call in tool_calls:
            tool = tools_by_name.get(tool_call.tool_name)
            additional_kwargs = {
                "tool_call_id": tool_call.tool_id,
                "name": tool.metadata.get_name(),
            }
            if not tool:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"Tool {tool_call.tool_name} does not exist",
                        additional_kwargs=additional_kwargs,
                    )
                )
                continue

            try:
                tool_output = tool(**tool_call.tool_kwargs)
                sources.append(tool_output)
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=tool_output.content,
                        additional_kwargs=additional_kwargs,
                    )
                )
            except Exception as e:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"Encountered error in tool call: {e}",
                        additional_kwargs=additional_kwargs,
                    )
                )

        # atualizar memória
        memory = await ctx.store.get("memory")
        for msg in tool_msgs:
            memory.put(msg)

        await ctx.store.set("sources", sources)
        await ctx.store.set("memory", memory)

        chat_history = memory.get()
        return InputEvent(input=chat_history)

Entendendo cada método

prepare_chat_history(): Ponto de entrada principal. Gerencia a adição da mensagem do usuário à memória e recupera o histórico completo de chat.

handle_llm_input(): Acionado por InputEvent, usa o histórico e ferramentas para consultar o LLM. Se tool calls forem encontradas, emite ToolCallEvent. Caso contrário, finaliza com StopEvent.

handle_tool_calls(): Acionado por ToolCallEvent, executa as ferramentas com tratamento de erros e retorna os resultados. Este evento cria um loop ao emitir um novo InputEvent, retornando para handle_llm_input().

Testando o agente

Vamos criar ferramentas simples de matemática e testar nosso agente:

from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

def add(x: int, y: int) -> int:
    """Função útil para somar dois números."""
    return x + y

def multiply(x: int, y: int) -> int:
    """Função útil para multiplicar dois números."""
    return x * y

tools = [
    FunctionTool.from_defaults(add),
    FunctionTool.from_defaults(multiply),
]

agent = FuncationCallingAgent(
    llm=OpenAI(model="gpt-4o-mini"), 
    tools=tools, 
    timeout=120, 
    verbose=True
)

ret = await agent.run(input="Hello!")
print(ret["response"])

Nota importante: Como estamos trabalhando com loops, precisamos definir um timeout adequado. Aqui configuramos 120 segundos.

Testando cálculos complexos

ret = await agent.run(input="What is (2123 + 2321) * 312?")

O agente irá:

1. Identificar que precisa fazer uma soma

2. Chamar a função add(2123, 2321)

3. Pegar o resultado (4444)

4. Identificar que precisa multiplicar

5. Chamar a função multiply(4444, 312)

6. Retornar o resultado final

Gerenciando histórico de conversas

Por padrão, o workflow cria um novo Context para cada execução, o que significa que o histórico não é preservado entre chamadas. Para manter continuidade na conversa, passe seu próprio Context:

from llama_index.core.workflow import Context

ctx = Context(agent)

ret = await agent.run(input="Hello! My name is Logan.", ctx=ctx)
print(ret["response"])
# Output: Hello, Logan! How can I assist you today?

ret = await agent.run(input="What is my name?", ctx=ctx)
print(ret["response"])
# Output: Your name is Logan.

Implementando streaming de respostas

Uma das funcionalidades mais interessantes é o streaming em tempo real. Usando o handler retornado pelo método .run(), podemos acessar os eventos de streaming:

agent = FuncationCallingAgent(
    llm=OpenAI(model="gpt-4o-mini"), 
    tools=tools, 
    timeout=120, 
    verbose=False
)

handler = agent.run(input="Hello! Write me a short story about a cat.")

async for event in handler.stream_events():
    if isinstance(event, StreamEvent):
        print(event.delta, end="", flush=True)

response = await handler

Isso permite criar interfaces de usuário mais interativas, onde o texto aparece gradualmente, melhorando a experiência do usuário.

Casos de uso práticos

Este padrão de agente com function calling pode ser adaptado para diversos cenários:

• Assistentes de código: ferramentas para executar código, buscar documentação, fazer debugging

• Análise de dados: funções para consultar databases, gerar gráficos, calcular estatísticas

• Automação de tarefas: integração com APIs externas, envio de emails, agendamento

• Pesquisa e síntese: busca em documentos, extração de informações, sumarização

Conclusão

Neste tutorial, construímos do zero um agente conversacional com function calling usando LlamaIndex Workflows. Aprendemos sobre:

• A arquitetura de um agente com function calling

• Como definir eventos customizados em workflows

• Implementação de loops e gerenciamento de estado

• Preservação de histórico de conversas

• Streaming de respostas em tempo real

O padrão de workflow apresentado é extremamente flexível e pode ser expandido com mais ferramentas, melhor tratamento de erros, e integrações com sistemas externos. A capacidade dos LLMs de usar ferramentas de forma inteligente abre possibilidades infinitas para criar aplicações verdadeiramente autônomas e úteis.

O código completo está disponível na documentação oficial do LlamaIndex, e você pode adaptá-lo para suas necessidades específicas. Experimente adicionar suas próprias ferramentas e veja como o agente aprende a usá-las de forma natural!