Uma Arquitetura de Agentes baseada em Grafos para a Resolução de Tarefas Complexas com LLMs
A crescente complexidade das aplicações de inteligência artificial exige soluções cada vez mais sofisticadas. Neste trabalho, apresentamos uma arquitetura baseada em múltiplos agentes de LLM, interconectados por um grafo, para lidar com a diversidade de tarefas e a complexidade das solicitações dos usuários. Ao dividir as tarefas em módulos especializados, conseguimos aumentar a eficiência, a escalabilidade e a flexibilidade do sistema, permitindo a criação de soluções personalizadas e adaptáveis.
LangGraph
O desenvolvimento de sistemas de inteligência artificial cada vez mais sofisticados exige ferramentas capazes de lidar com a complexidade crescente dessas aplicações. O LangGraph surge como uma resposta a essa demanda, oferecendo uma plataforma flexível e intuitiva para a construção de sistemas de IA personalizados. Sua representação em grafo permite modelar de forma clara as relações entre diferentes componentes de um sistema, facilitando a compreensão e a manutenção. Além disso, a integração nativa com modelos de linguagem de grande porte (LLMs) torna o LangGraph uma ferramenta ideal para a criação de assistentes virtuais inteligentes e chatbots personalizados. Ao permitir a criação de fluxos de trabalho complexos e a integração de diversos módulos, o LangGraph democratiza o desenvolvimento de IA, tornando-o acessível a um público mais amplo.
Escopo
Este artigo demonstra um fluxo de trabalho que permite a um usuário solicitar e visualizar dados de um banco de dados em diferentes formatos, como gráficos e tabelas. Utilizando a mesma base de dados do estudo anterior, detalhado em [link para o artigo anterior], exploramos a implementação de um sistema que atenda às necessidades de visualização personalizadas do usuário.
Agora vamos pôr a mão na massa.
Abaixo segue as bibliotecas que serão utilizadas.
from langchain_experimental.tools import PythonREPLTool
from langchain_community.agent_toolkits import create_sql_agent
from langchain_groq import ChatGroq
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain.callbacks.base import BaseCallbackHandler
from langchain_community.utilities import SQLDatabase
from IPython.display import Markdown as md
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Annotated, Sequence, TypedDict
import re
import os
import operator
import functoolsCom o objetivo de otimizar a identificação e resolução de problemas, contamos com o LangSmith. Essa ferramenta monitora todas as etapas das execuções do Langchain, permitindo a detecção precoce de anomalias e a análise detalhada do desempenho. Além disso, a disponibilidade de um plano gratuito torna a ferramenta acessível para projetos de menor escala.
os.environ["LANGCHAIN_TRACING_V2"] = "false"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "[token]"
os.environ["LANGCHAIN_PROJECT"] = "pr-teste-lang-graph"pyInstanciamos o modelo de linguagem de grande porte (LLM) que será utilizado nas tarefas de inteligência artificial generativa. A vantagem de usar múltiplos agentes é que cada um pode ter um modelo especializado, otimizado para tarefas específicas, permitindo um desempenho superior em cada área.
llm = ChatGroq(temperature=0, groq_api_key="[token]", model_name="llama3-70b-8192")Apresentamos a seguir a definição da classe de estado do agente, responsável por gerenciar e registrar as transições entre os estados de cada nó. Posteriormente, demonstraremos como essa classe é aplicada na prática.
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]O método a seguir será utilizado para criar uma instância de um agente. Este método só não será utilizado na criação do agente de SQL que já possui o seu próprio método de criação de instância.
# function that returns AgentExecutor with given tool and prompt
def create_agent(llm, tools: list, system_prompt: str):
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
system_prompt,
),
MessagesPlaceholder(variable_name="messages"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
return executorO próximo método tem como objetivo ser a execução quando o fluxo chega a um determinado nó. Exemplo: O agente de SQL executa a query e é redirecionado para a criação de um gráfico, quando isso ocorrer esse método será invocado através do nó da criação do gráfico via Python.
def agent_node(state, agent, name):
result = agent.invoke(state)
print(f'Execução do segundo agente: {str(result)}')
return {"messages": [HumanMessage(content=result["output"], name=name)]}Já o método de execução do SQL agente é único, pois foi necessário alguns ajustes na execução pois gostaria de devolver apenas os dados.
Obs.: Acredito que poderia ter utilizado outras abordagens, como por exemplo a customização do prompt para isso, mas como o intuito é mais acadêmico e mostrar o funcionamento acabei não me aprofundando nesse ponto.
class SQLHandler(BaseCallbackHandler):
def __init__(self):
self.sql_result = []
def on_agent_action(self, action, **kwargs):
"""Run on agent action. if the tool being used is sql_db_query,
it means we're submitting the sql and we can
record it as the final sql"""
if action.tool in ["sql_db_query"]:
self.sql_result.append(action.tool_input)
def agent_sql_node(state, agent, name):
handler = SQLHandler()
result = agent.invoke({'input': state["messages"][-1].content}, {'callbacks': [handler]})
if len(handler.sql_result) > 0:
sql_query = handler.sql_result[0]['query']
return {"messages": [SystemMessage(content="This is the return data from the database: " +
db.run(sql_query, fetch="all", include_columns=True), name=name)]}Este método atua como um ponto de decisão, encaminhando o fluxo para a próxima etapa com base na classificação extraída da solicitação do usuário por um modelo de linguagem de grande porte (LLM). As possíveis próximas etapas incluem a geração de gráficos, a criação de tabelas ou a resposta em linguagem natural.
def where_to_go(state):
messages = state['messages']
print(f'Mensagem enviada: {messages[0]}')
result = llm.invoke("based on user input:{}, classify whether the text intends for the return to be a chart, table or other if the internalization is not clear.Importantly, return only the text: [chart, table, other]".format(messages[0]))
print(f"Classe retornada na analise: {result.content}.")
if "chart" in result.content.lower():
return "chart"
if "table" in result.content.lower():
return "table"
else:
return "write"O código abaixo cria uma instância da ferramenta que será utilizada para executar o código Python e gerar o gráfico de acordo com a solicitação do usuário.
python_repl_tool = PythonREPLTool()Criação da conexão com o banco de dados para a execução das consultas geradas pelo LLM.
maria_uri = 'mysql+mysqlconnector://root:art_llama3@localhost:3306/mme'
db = SQLDatabase.from_uri(maria_uri)Segue abaixo a criação da estrutura de grafo da execução dos múltiplos agentes com o LangGraph.
# QueryBuild as a node
sql_agent = create_sql_agent(llm, db=db, agent_type="tool-calling", max_iterations=5, verbose=True, format_instructions="Return only raw data in JSON")
sql_node = functools.partial(agent_sql_node, agent=sql_agent, name="QueryBuild")
# Nodes
code_agent = create_agent(llm, [python_repl_tool], "You generate charts using matplotlib.")
code_node = functools.partial(agent_node, agent=code_agent, name="Coder")
table_agent = create_agent(llm, [], "You generate markdown using data from database.")
table_node = functools.partial(agent_node, agent=table_agent, name="Table")
write_agent = create_agent(llm, [], "Write a natural language response only in portuguese(pt-BR) from data of the database.")
write_node = functools.partial(agent_node, agent=write_agent, name="Write")
# defining the StateGraph
workflow = StateGraph(AgentState)
workflow.add_node("QueryBuild", sql_node)
workflow.add_node("Coder", code_node)
workflow.add_node("Table", table_node)
workflow.add_node("Write", write_node)
workflow.add_conditional_edges("QueryBuild", where_to_go, { # Based on the return from where_to_go
# If return is "chart", "table" or "write" then we call the tool node.
"chart": "Coder",
"table": "Table",
"write": "Write",
# Otherwise we finish. END is a special node marking that the graph should finish.
"end": END
}
)
# starting point should be QueryBuild
workflow.set_entry_point("QueryBuild")
workflow.set_finish_point("Coder")
workflow.set_finish_point("Table")
workflow.set_finish_point("Write")
graph = workflow.compile()Abaixo é possivel ver a representação gráfica do grafo criado para a execução, onde existe 4 agentes distintos.
print(graph.get_graph().draw_ascii())
Abaixo é realizada a execução do primeiro teste dos múltiplos agentes com a criação de um gráfico no final da execução.
final_state = graph.invoke(
{"messages": [HumanMessage(content="Qual foi o top 4 estados que tiveram o maior consumo em MWh total? Gere um gráfico de barras com o estado e o total MWh")]}
)> Entering new SQL Agent Executor chain...
Invoking: `sql_db_list_tables` with `{'tool_input': ''}`
consumo_energia_eletrica, uf
Invoking: `sql_db_schema` with `{'table_names': 'consumo_energia_eletrica, uf'}`
CREATE TABLE consumo_energia_eletrica (
ano INTEGER(11),
mes INTEGER(11),
sigla_uf VARCHAR(255),
tipo_consumo VARCHAR(255),
numero_consumidores INTEGER(11),
`consumo_MWh` FLOAT,
CONSTRAINT `sigla_uf_FK` FOREIGN KEY(sigla_uf) REFERENCES uf (sigla)
)ENGINE=InnoDB COLLATE utf8mb4_general_ci DEFAULT CHARSET=utf8mb4
/*
3 rows from consumo_energia_eletrica table:
ano mes sigla_uf tipo_consumo numero_consumidores consumo_MWh
2004 1 RO Residencial 258610 44271.0
2004 1 AC Residencial 103396 15778.1
2004 1 AM Residencial 480619 84473.0
*/
CREATE TABLE uf (
sigla VARCHAR(2) NOT NULL,
nome_do_estado VARCHAR(255),
PRIMARY KEY (sigla)
)ENGINE=InnoDB COLLATE utf8mb4_general_ci DEFAULT CHARSET=utf8mb4
/*
3 rows from uf table:
sigla nome_do_estado
AC Acre
AL Alagoas
AM Amazonas
*/
Invoking: `sql_db_query` with `{'query': 'SELECT uf.nome_do_estado, SUM(consumo_MWh) as total_MWh FROM consumo_energia_eletrica JOIN uf ON consumo_energia_eletrica.sigla_uf = uf.sigla GROUP BY uf.nome_do_estado ORDER BY total_MWh DESC LIMIT 4'}`
[('São Paulo', 2247429685.375), ('Minas Gerais', 934651225.28125), ('Rio de Janeiro', 661774178.34375), ('Paraná', 492105686.1875)]The top 4 states with the highest total MWh consumption are:
1. São Paulo - 2247429685.375 MWh
2. Minas Gerais - 934651225.28125 MWh
3. Rio de Janeiro - 661774178.34375 MWh
4. Paraná - 492105686.1875 MWh
Here is a bar chart to illustrate the results:
```
São Paulo |************************************************************
Minas Gerais |***********************************
Rio de Janeiro |*******************************
Paraná |***************************
```
Let me know if you need anything else!
> Finished chain.
Mensagem enviada: content='Qual foi o top 4 estados que tiveram o maior consumo em MWh total? Gere um gráfico de barras com o estado e o total MWh'
Classe retornada na analise: chart.
Execução do segundo agente: {'messages': [HumanMessage(content='Qual foi o top 4 estados que tiveram o maior consumo em MWh total? Gere um gráfico de barras com o estado e o total MWh'), SystemMessage(content="This is the return data from the database: [{'nome_do_estado': 'São Paulo', 'total_MWh': 2247429685.375}, {'nome_do_estado': 'Minas Gerais', 'total_MWh': 934651225.28125}, {'nome_do_estado': 'Rio de Janeiro', 'total_MWh': 661774178.34375}, {'nome_do_estado': 'Paraná', 'total_MWh': 492105686.1875}]", name='QueryBuild')], 'next': None, 'output': 'O gráfico de barras foi gerado com sucesso, mostrando os 4 estados com o maior consumo em MWh total.'}Abaixo é realizada a execução do segundo teste dos múltiplos agentes com a criação de uma tabela no final da execução.
final_state = graph.invoke(
{"messages": [HumanMessage(content="Quais foram o consumo em MWh total nos anos 2004, 2005 e 2006 para os estados SP, RJ, GO e DF? Crei uma tabela com as colunas para cada ano e uma linha para cada estado.")]}
)> Entering new SQL Agent Executor chain...
Invoking: `sql_db_list_tables` with `{'tool_input': ''}`
consumo_energia_eletrica, uf
Invoking: `sql_db_schema` with `{'table_names': 'consumo_energia_eletrica, uf'}`
CREATE TABLE consumo_energia_eletrica (
ano INTEGER(11),
mes INTEGER(11),
sigla_uf VARCHAR(255),
tipo_consumo VARCHAR(255),
numero_consumidores INTEGER(11),
`consumo_MWh` FLOAT,
CONSTRAINT `sigla_uf_FK` FOREIGN KEY(sigla_uf) REFERENCES uf (sigla)
)ENGINE=InnoDB COLLATE utf8mb4_general_ci DEFAULT CHARSET=utf8mb4
/*
3 rows from consumo_energia_eletrica table:
ano mes sigla_uf tipo_consumo numero_consumidores consumo_MWh
2004 1 RO Residencial 258610 44271.0
2004 1 AC Residencial 103396 15778.1
2004 1 AM Residencial 480619 84473.0
*/
CREATE TABLE uf (
sigla VARCHAR(2) NOT NULL,
nome_do_estado VARCHAR(255),
PRIMARY KEY (sigla)
)ENGINE=InnoDB COLLATE utf8mb4_general_ci DEFAULT CHARSET=utf8mb4
/*
3 rows from uf table:
sigla nome_do_estado
AC Acre
AL Alagoas
AM Amazonas
*/
Invoking: `sql_db_query` with `{'query': "SELECT uf.sigla, SUM(consumo_MWh) as consumo_MWh_2004, SUM(CASE WHEN ano = 2005 THEN consumo_MWh ELSE 0 END) as consumo_MWh_2005, SUM(CASE WHEN ano = 2006 THEN consumo_MWh ELSE 0 END) as consumo_MWh_2006 FROM consumo_energia_eletrica INNER JOIN uf ON consumo_energia_eletrica.sigla_uf = uf.sigla WHERE uf.sigla IN ('SP', 'RJ', 'GO', 'DF') GROUP BY uf.sigla"}`
[('DF', 102556716.65039062, 4102913.466796875, 4309621.83984375), ('GO', 229683315.5, 8472150.15625, 8615403.28125), ('RJ', 661774178.34375, 32227671.9375, 32393585.0625), ('SP', 2247429685.375, 103906718.0, 109148432.8125)]The answer is:
| Estado | 2004 | 2005 | 2006 |
| --- | --- | --- | --- |
| DF | 102556716.65 | 4102913.47 | 4309621.84 |
| GO | 229683315.5 | 8472150.16 | 8615403.28 |
| RJ | 661774178.34 | 32227671.94 | 32393585.06 |
| SP | 2247429685.38 | 103906718.0 | 109148432.81 |
Let me know if you need anything else!
> Finished chain.
Mensagem enviada: content='Quais foram o consumo em MWh total nos anos 2004, 2005 e 2006 para os estados SP, RJ, GO e DF? Crei uma tabela com as colunas para cada ano e uma linha para cada estado.'
Classe retornada na analise: Based on the user input, I would classify the intended return as: table.
Execução do segundo agente: {'messages': [HumanMessage(content='Quais foram o consumo em MWh total nos anos 2004, 2005 e 2006 para os estados SP, RJ, GO e DF? Crei uma tabela com as colunas para cada ano e uma linha para cada estado.'), SystemMessage(content="This is the return data from the database: [{'sigla': 'DF', 'consumo_MWh_2004': 102556716.65039062, 'consumo_MWh_2005': 4102913.466796875, 'consumo_MWh_2006': 4309621.83984375}, {'sigla': 'GO', 'consumo_MWh_2004': 229683315.5, 'consumo_MWh_2005': 8472150.15625, 'consumo_MWh_2006': 8615403.28125}, {'sigla': 'RJ', 'consumo_MWh_2004': 661774178.34375, 'consumo_MWh_2005': 32227671.9375, 'consumo_MWh_2006': 32393585.0625}, {'sigla': 'SP', 'consumo_MWh_2004': 2247429685.375, 'consumo_MWh_2005': 103906718.0, 'consumo_MWh_2006': 109148432.8125}]", name='QueryBuild')], 'next': None, 'output': 'Aqui está a tabela em markdown com os dados solicitados:\n\n| Estado | 2004 | 2005 | 2006 |\n| --- | --- | --- | --- |\n| DF | 102556716.65 | 4102913.47 | 4309621.84 |\n| GO | 229683315.5 | 8472150.16 | 8615403.28 |\n| RJ | 661774178.34 | 32227671.94 | 32393585.06 |\n| SP | 2247429685.38 | 103906718.0 | 109148432.81 |\n\nNote que os valores estão arredondados para 2 casas decimais. Se você precisar de mais precisão, basta me informar!'}md(final_state['messages'][-1].content)
Abaixo é realizada a execução do terceiro teste dos múltiplos agentes apenas respondendo em linguagem natural no final da execução.
final_state = graph.invoke(
{"messages": [HumanMessage(content="Qual foi o consumo em MWh total em 2004?")]}
)> Entering new SQL Agent Executor chain...
Invoking: `sql_db_list_tables` with `{'tool_input': ''}`
consumo_energia_eletrica, uf
Invoking: `sql_db_schema` with `{'table_names': 'consumo_energia_eletrica'}`
CREATE TABLE consumo_energia_eletrica (
ano INTEGER(11),
mes INTEGER(11),
sigla_uf VARCHAR(255),
tipo_consumo VARCHAR(255),
numero_consumidores INTEGER(11),
`consumo_MWh` FLOAT,
CONSTRAINT `sigla_uf_FK` FOREIGN KEY(sigla_uf) REFERENCES uf (sigla)
)ENGINE=InnoDB COLLATE utf8mb4_general_ci DEFAULT CHARSET=utf8mb4
/*
3 rows from consumo_energia_eletrica table:
ano mes sigla_uf tipo_consumo numero_consumidores consumo_MWh
2004 1 RO Residencial 258610 44271.0
2004 1 AC Residencial 103396 15778.1
2004 1 AM Residencial 480619 84473.0
*/
Invoking: `sql_db_query` with `{'query': 'SELECT SUM(consumo_MWh) AS total_consumo FROM consumo_energia_eletrica WHERE ano = 2004'}`
[(331865173.9071045,)]The total consumption in MWh in 2004 is 331865173.91 MWh.
> Finished chain.
Mensagem enviada: content='Qual foi o consumo em MWh total em 2004?'
Classe retornada na analise: Based on the user input, I would classify the text as intending for the return to be a numerical value, so I would return:
other.
Execução do segundo agente: {'messages': [HumanMessage(content='Qual foi o consumo em MWh total em 2004?'), SystemMessage(content="This is the return data from the database: [{'total_consumo': 331865173.9071045}]", name='QueryBuild')], 'next': None, 'output': 'De acordo com os dados, o consumo total em MWh foi de 331.865.173,91 MWh.'}final_state['messages'][-1].contentDe acordo com os dados, o consumo total em MWh foi de 331.865.173,91 MWh.
O código desse notebook pode ser baixado aqui.
Conclusão
Em resumo, este trabalho demonstrou a eficácia do LangGraph na criação de sistemas de processamento de dados personalizados e escaláveis. A modularidade do grafo e a facilidade de configuração permitiram a rápida implementação de um sistema capaz de atender a diversas solicitações dos usuários. Os testes realizados evidenciaram a capacidade do sistema de seguir caminhos distintos de execução, de acordo com as especificações de cada solicitação. Embora este trabalho tenha se concentrado em um caso de uso específico, os resultados obtidos demonstram o potencial do LangGraph para a construção de sistemas de inteligência artificial mais complexos e sofisticados.
