Além do bot: A anatomia de um agente operacional para emergências
Quando se fala em automação em geoprocessamento, a imagem mais comum é a de um script rodando em loop ou de um bot que responde a comandos pré-definidos. Em cenários críticos, como acidentes de trânsito, essa rigidez pode virar risco: o sistema precisa lidar com incerteza, falhas e mudanças de contexto. Por isso, mais do que um executor de tarefas, o que faz diferença é um Agent.
A distinção é simples, mas é bastante importante. Enquanto um bot percorre um fluxo linear, um Agent interpreta o contexto, escolhe quais tools acionar, em que ordem, e ajusta o plano conforme a informação disponível, para esse caso específico ele não apenas reage, ele coordena o processo.
Neste artigo, vou destrinchar a construção de um operational GIS Agent, desenvolvido para situações de emergência. A ideia é explicar o raciocínio por trás do sistema, porque cada parte existe e como decisões de projeto transformam um pipeline de análise espacial em uma aplicação interativa.
O resultado esperado é simples:
- Localizar a ocorrência usando pontos de referência;
- Encontrar o hospital mais próximo;
- Traçar uma rota no mapa e gerar uma mensagem modelo de comunicação.
Preparando o terreno: dependências e constantes
Antes de falar em autonomia, o ambiente precisa estar pronto para executar ações “de verdade”. Isso exige três coisas básica, encontrar uma base espacial confiável, integrar os serviços externos e uma interface que sustente uso interativo.
Na camada espacial, GeoPandas e Shapely permitem manipular geometrias e fazer operações como interseção e recorte e outros processos de geoprocessamento. Mesmo quando o usuário só vê um ponto no mapa, por trás existe um conjunto de cálculos que precisa ser realizado, então é aí que entra nossos amigos NumPy e pandas entram como suporte para dados numéricos e estruturas tabulares, especialmente quando o resultado precisa de agregações e conversões. Já para integração com dados externos, usamos a biblioteca requests que faz a conexão nas APIs públicas onde iremos adquirir os dados.
Na interface, a combinação de ipywidgets com ipyleaflet para transforma um notebook em uma aplicação simples, mas interativa. Esse tipo de abordagem funciona bem em prova de conceito e permite a gente explorar o fluxo sem precisar construir um front-end completo, o que foi ideal para mim, já que não tenho muita familiaridade com desenvolvimento de interface.
import os, json, requests, polyline as pl
import numpy as np
import pandas as pd
import geopandas as gpd
import time
import random
from shapely.geometry import Point, LineString
from shapely.ops import transform
from pyproj import CRS, Transformer
import ipywidgets as w
from ipyleaflet import (
Map, Marker, Polyline, Polygon as LeafPoly,
GeoJSON, LayerGroup, AwesomeIcon, basemaps, basemap_to_tiles,
CircleMarker, DivIcon
)
from IPython.display import display
Também existem parâmetros fixos importantes. O OVERPASS_URL aponta para a Overpass API, usada para consultar dados do OpenStreetMap, que vão nos retornar informações no entorno do das ocorrências (vias, hospitais e referências). Já o OSRM_URL define o serviço de roteamento (OSRM), que vamos usar para calcular a melhor rota até o hospital mais próximo, incluindo distância, duração e a própria geometria da rota.
Outras definições importantes é o DEFAULT_BUFFER_M define o raio padrão de análise, quanto de dados vamos trazer da API para analisar, o qual temos que tentar equilibrar a definição do contexto, mas se alertando ao desempenho, para evitar baixar e analisar a cidade toda. Em seguida precisamos definir DEFAULT_ROAD_TAGS delimita quais classes de vias são relevantes para a resposta, neste caso para evitar trazer contexto desnecessário para a aplicação que não irá impactar na ocorrência. Por fim, entram as configurações da API de IA, neste caso, a da OpenAI.
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
OSRM_URL = "https://router.project-osrm.org/route/v1/driving"
DEFAULT_BUFFER_M = 2000
DEFAULT_ROAD_TAGS = ["motorway","trunk","primary","secondary",
"tertiary","residential","service"]
# OpenAI
from openai import OpenAI
from google.colab import userdata
os.environ["OPENAI_API_KEY"]=userdata.get('OPENAI_API_KEY')
client = OpenAI()
Projeção local: buffer em metros e consistência de medida
Um erro comum em análises espaciais rápidas é tratar latitude/longitude como unidade métrica, isso é muito comum para quem nunca trabalhou com Geo. Para visualização, podemos utilizar tranquilamente, mas para medida, não é o ideal. Quando calculamos distância, comprimento de via e buffer precisam ser calculados em CRS projetado, caso contrário, o resultado varia com a latitude e pode distorcer toda a análise.
Para evitar isso vamos definir uma função auxiliar calcula automaticamente um CRS UTM local, assim, podemos garantir a portabilidade das feições e realizar a projeção sem depender de uma zona fixa. No final o fluxo é bem simples, a coordenada entra em WGS84, é transformada para UTM para aplicar o buffer em metros, e depois retorna para WGS84 para manter compatibilidade.
def local_utm_crs(lon, lat):
zone = int((lon + 180) / 6) + 1
return CRS.from_epsg((32700 if lat < 0 else 32600) + zone)
def make_buffer_m(lat, lon, buf):
wgs = CRS.from_epsg(4326)
utm = local_utm_crs(lon, lat)
to_utm = Transformer.from_crs(wgs, utm, always_xy=True).transform
to_wgs = Transformer.from_crs(utm, wgs, always_xy=True).transform
return transform(to_wgs, transform(to_utm, Point(lon, lat)).buffer(buf))
Overpass* com mirrors e tolerância a falhas
Outro ponto importante para ser alertado é que em um pipeline acadêmico, falha de API é inconveniente, já em um pipeline operacional, é esperado. Por isso, é de extrema importância tratar a instabilidade como parte do cenário.
A estratégia para resolver esse problema é combina três decisões: uso de múltiplos mirrors (reduzindo dependência de um único endpoint), tentativas com backoff exponencial (evitando insistência agressiva) e tratamento explícito de rate limit, especialmente quando a API devolve 429. Esse conjunto reduz quedas e aumenta a chance de resposta útil em condições reais.
OVERPASS_MIRRORS = [
"https://overpass-api.de/api/interpreter",
"https://overpass.kumi.systems/api/interpreter",
"https://maps.mail.ru/osm/tools/overpass/api/interpreter",
"https://overpass.openstreetmap.ru/api/interpreter",
"https://overpass.osm.ch/api/interpreter",
]
def overpass_query(q, timeout=60, retries=3):
for mirror in OVERPASS_MIRRORS:
for attempt in range(1, retries + 1):
try:
print(f" → Overpass: {mirror} (tentativa {attempt})")
r = requests.post(mirror, data={"data": q}, timeout=timeout)
r.raise_for_status()
return r.json()
except requests.exceptions.Timeout:
wait = (2 ** attempt) + random.uniform(0, 1) # backoff exponencial
print(f" Timeout, aguardando {wait:.1f}s…")
time.sleep(wait)
except requests.exceptions.HTTPError as e:
if r.status_code == 429: # rate limit explícito
wait = int(r.headers.get("Retry-After", 30))
print(f" Rate limit (429), aguardando {wait}s…")
time.sleep(wait)
continue
if r.status_code in (502, 503, 504):
time.sleep(2 ** attempt)
break
raise
except requests.exceptions.ConnectionError:
break
raise RuntimeError("Overpass indisponível em todos os mirrors.")
Do OSM para o GIS: organizar antes de analisar
A resposta da Overpass mistura o que vem da API, informações como nodes, ways e relations no mesmo conjunto. Para um Agente que estamos contruindo se trabalharmos com tudo isso torna, pode tornar o pipeline frágil e difícil de manter, pelo menos foi isso que observei em outros testes. O primeiro passo, então, é organizar os dados em estruturas previsíveis.
A função de parsing separa exatamente isso: cria um dicionário de nodes por ID para acesso rápido e listas dedicadas para ways e relations, gerando o dado final menos complexo e as etapas seguintes ficam mais estáveis.
def parse_elements(elements):
nodes, ways, rels = {}, [], []
for el in elements:
t = el.get("type")
if t == "node": nodes[el["id"]] = el
elif t == "way": ways.append(el)
elif t == "relation": rels.append(el)
return nodes, ways, rels
Ainda sobre os dados OpenStreetMap, é importante também tratar que uma via (way) não chega como uma linha pronta, ela vem como uma lista de IDs de nós. A conversão para geometria exige resolver esses IDs em coordenadas e, só então, montar uma LineString, que possibilita recortar, medir e visualizar as vias com maior facilidade.
def way_to_ls(way, nodes):
coords = [(nodes[n]["lon"], nodes[n]["lat"])
for n in way.get("nodes", []) if n in nodes]
return LineString(coords) if len(coords) >= 2 else None
Entretanto emergência, mais dados não significa melhor resposta, então, o excesso de informação pode confundir o modelo e até induzir uma resposta diferente do esperado, então para facilitar nosso trabalho precisamos filtrar somente as vias relevantes para deslocamento e contexto operacional, não de todas as trilhas e tags marginais que possam aparecer. Por isso, criamos uma função auxiliar que filtra ways pela tag highway e, quando há uma lista de classes, mantém apenas categorias definidas como úteis em DEFAULT_ROAD_TAGS.
def extract_roads(nodes, ways, road_tags=None):
rt = set(road_tags or [])
rows = []
for w in ways:
tags = w.get("tags", {}); hw = tags.get("highway")
if not hw or (rt and hw not in rt): continue
geom = way_to_ls(w, nodes)
if geom: rows.append({"osm_id": w["id"], "highway": hw,
"name": tags.get("name"), "geometry": geom})
return (gpd.GeoDataFrame(rows, geometry="geometry", crs="EPSG:4326") if rows
else gpd.GeoDataFrame(columns=["osm_id","highway","name","geometry"],
geometry="geometry", crs="EPSG:4326"))
Sabemos que os dados do OpenStreetMap (OSM) são bastante ricos, mas nem sempre consistentes, já que são levantados pela própria comunidade. Por exemplo, um hospital pode ter o campo name, pode ter apenas operator, pode não ter nenhum desses atributos ou até existir apenas como um endereço.
Para o agente, isso é um ponto crítico, pois a utilidade do resultado final depende da capacidade de nomear destinos e referências de forma clara e compreensível. Imagine se todas as referências repassadas à equipe do SAMU estivessem sem as informações necessárias.
NAME_FIELDS = ["name", "operator"]
POI_NAME_FIELDS = ["name", "brand"]
POI_KIND_FIELDS = ["amenity", "tourism", "leisure", "shop"]
def get_name(tags: dict) -> str:
named = next((tags[f] for f in NAME_FIELDS if tags.get(f)), None)
if named:
return named
street = tags.get("addr:street", "")
number = tags.get("addr:housenumber", "")
address = f"{street}, {number}".strip(", ")
return address or "Hospital sem nome cadastrado"
Continuando sobre a política de nomeação, essa define uma ordem de prioridade e aplica fallback ("Hopital sem nome cadastrado") para endereço quando necessário. Isso evita respostas genéricas e melhora a clareza.
A partir daí entram duas funções auxiliares:
- get_poi_name, que percorre os campos definidos em POI_NAME_FIELDS e retorna o primeiro valor disponível, caso contrário retorna “Ponto de Referência”
- get_poi_kind, que complementa a identificação classificando o tipo do ponto de referência
def get_poi_name(tags: dict) -> str:
return next((tags[f] for f in POI_NAME_FIELDS if tags.get(f)), "Ponto de referência")
def get_poi_kind(tags: dict) -> str:
return next((tags[k] for k in POI_KIND_FIELDS if tags.get(k)), "local")
Outro ponto que precisa ser considerado é que os hospitais podem aparecer como node, way ou relation. Por isso, é essencial contemplar essas possibilidades, reduzindo falsos negativos, como casos em que o sistema indica “hospital não encontrado” apenas por causa da forma como o dado foi modelado.
Além disso, ao padronizar os nomes, o agente tende a produzir um texto final mais útil e legível.
def extract_hospitals(nodes, ways, rels):
rows = []
for n in nodes.values():
tags = n.get("tags", {})
if tags.get("amenity") == "hospital":
rows.append({"name": get_name(tags), "geometry": Point(n["lon"], n["lat"])})
for w in ways:
tags = w.get("tags", {})
if tags.get("amenity") == "hospital":
g = way_to_ls(w, nodes)
if g:
rows.append({"name": get_name(tags), "geometry": g.centroid})
for r in rels:
tags = r.get("tags", {})
if tags.get("amenity") == "hospital" and "center" in r:
c = r["center"]
rows.append({"name": get_name(tags), "geometry": Point(c["lon"], c["lat"])})
cols = ["name", "geometry"]
return (gpd.GeoDataFrame(rows, geometry="geometry", crs="EPSG:4326")
if rows else gpd.GeoDataFrame(columns=cols, geometry="geometry", crs="EPSG:4326"))
Mesmo com uma coordenada precisa, como comentei no artigo sobre H3, o par lat/lon, por si só, não ajuda muito a IA a entender o contexto. Para esse tipo de caso, é bem mais efetivo trabalhar com pontos de referência.
Por isso, criei uma função que extrai e busca elementos com tags de interesse (serviços, comércio, turismo e lazer), remove itens que podem confundir o relato, como hospitais, que já foram extraídos anteriormente, e retorna dois candidatos, priorizando a proximidade. Assim, fica mais fácil para a IA, ou até mesmo para o SAMU, entender que o acidente ocorreu perto da Loja X e do Shopping Y. Se quiser deixar isso ainda mais completo, poderiamos até incluir o endereço usando o reverse geocoding via https://nominatim.openstreetmap.org/reverse.
POI_TAGS = {"amenity", "shop", "tourism", "leisure"}
def extract_reference_points(nodes, lat, lon, max_points: int = 2):
candidates = []
for n in nodes.values():
tags = n.get("tags", {})
if not POI_TAGS.intersection(tags) or tags.get("amenity") == "hospital":
continue
dist_m = ((n["lat"] - lat) ** 2 + (n["lon"] - lon) ** 2) ** 0.5 * 111_000
candidates.append({
"name": get_poi_name(tags),
"type": get_poi_kind(tags),
"lat": n["lat"],
"lon": n["lon"],
"dist_m": dist_m,
})
candidates.sort(key=lambda x: x["dist_m"])
return candidates[:max_points]
Análise de impacto no entorno do acidente
Com o buffer, a consulta ao Overpass e a conversão para geometrias, o Agent passa a ter contexto do entorno do ocorrido usando a função compute_impact. Ela começa montando uma única consulta, que busca roads, hospitals e points of interest dentro do mesmo envelope espacial (região de busca), e recorta as vias para considerar apenas o que de fato cai dentro do buffer.
Em seguida, o resultado é reprojetado para um sistema em unidades métricas. A partir daí, calcula-se o comprimento das vias e faz-se a agregação por classe, o que dá uma noção de densidade e do tipo de infraestrutura no entorno. Em paralelo, a distância até o hospital mais próximo também é calculada.
Tudo isso volta como contexto do acidente em um JSON, com informações como exemplo: accident_location, radius, road_count, total_road_length, road_length_by_class, nearest_hospital e landmarks.
def compute_impact(lat, lon, buf, road_tags=None):
area = make_buffer_m(lat, lon, buf)
s, w2, n, e = area.bounds[1], area.bounds[0], area.bounds[3], area.bounds[2]
q = f"""[out:json][timeout:180];
(way["highway"]({s},{w2},{n},{e});
node["amenity"="hospital"]({s},{w2},{n},{e});
way["amenity"="hospital"]({s},{w2},{n},{e});
relation["amenity"="hospital"]({s},{w2},{n},{e});
node["amenity"~"pharmacy|fuel|police|fire_station|bank"]({s},{w2},{n},{e});
node["shop"~"supermarket|bakery|convenience"]({s},{w2},{n},{e});
);(._;>;);out body center;"""
data = overpass_query(q)
nodes, ways, rels = parse_elements(data.get("elements", []))
roads_gdf = extract_roads(nodes, ways, road_tags or DEFAULT_ROAD_TAGS)
hosp_wgs = extract_hospitals(nodes, ways, rels)
roads_in = roads_gdf[roads_gdf.intersects(area)].copy()
utm = local_utm_crs(lon, lat)
roads_utm = roads_in.to_crs(utm)
hosp_utm = hosp_wgs.to_crs(utm)
pt_utm = (gpd.GeoDataFrame([{"geometry": Point(lon, lat)}], crs="EPSG:4326")
.to_crs(utm).iloc[0].geometry)
total_m, by_class = 0.0, {}
if len(roads_utm):
roads_utm["len_m"] = roads_utm.length
total_m = float(roads_utm["len_m"].sum())
by_class = (roads_utm.groupby("highway")["len_m"].sum()
.sort_values(ascending=False).to_dict())
nearest = None
if len(hosp_utm):
hosp_utm["dist_m"] = hosp_utm.distance(pt_utm)
idx = hosp_utm["dist_m"].idxmin()
hw = hosp_wgs.loc[idx]
nearest = {
"name": hw.get("name"),
"dist_m": float(hosp_utm.loc[idx, "dist_m"]),
"lat": float(hw.geometry.y),
"lon": float(hw.geometry.x),
}
ref_points = extract_reference_points(nodes, lat, lon, max_points=2)
result = {
"accident": {"lat": lat, "lon": lon},
"buffer_m": buf,
"roads_count": int(len(roads_in)),
"roads_total_len_m": float(total_m),
"roads_len_by_class_m": {k: float(v) for k, v in by_class.items()},
"nearest_hospital": nearest,
"reference_points": ref_points,
}
return result, area, roads_in, hosp_wgs
Rota até o hospital mais próximo
Depois que o Agent identifica o nearest hospital, o resultado deixa de ser apenas contexto e passa a orientar uma ação. A etapa que acontece em paralelo é a de roteirização, que transforma um destino em um caminho navegável, incluindo distância, duração e geometria da rota.
Essa roteirização é feito com uma chamada à API pública do OSRM em https://router.project-osrm.org/route/v1/driving, informando origem e destino no formato lon,lat. A resposta traz uma ou mais rotas e, aqui, usamos a primeira, extraindo:
- distance_m, distância total em metros
- duration_s, duração estimada em segundos
- geometry, o traçado completo da rota, decodificado de polyline para uma lista de coordenadas
def get_route(olat, olon, dlat, dlon):
url = f"{OSRM_URL}/{olon},{olat};{dlon},{dlat}"
r = requests.get(url, params={"overview":"full","geometries":"polyline"},
timeout=30)
r.raise_for_status()
d = r.json()
if d.get("code") != "Ok" or not d.get("routes"):
raise ValueError(f"OSRM sem rota: {d}")
rt = d["routes"][0]
return {"distance_m": rt["distance"], "duration_s": rt["duration"],
"geometry": pl.decode(rt["geometry"])}
Comunicação: gerando uma mensagem modelo de chamado
Aqui é um ponto interessante, em um ambiente de produção, a função draft_samu_message ficaria aberta para evolução como um componente dedicado e isolado, já que comunicação é uma etapa sensível e costuma exigir padronização, rastreabilidade e auditoria. A ideia seria que ela não apenas montasse a mensagem, mas estruturasse o chamado para envio aos órgãos competentes, no formato e no canal definidos pela operação, por exemplo, integração via API, fila de mensagens ou sistema interno.
Aqui, como o projeto é exemplificado, então essa etapa permanece integrada ao tool executor para demonstrar o fluxo completo de ponta a ponta sem aumentar a complexidade estrutural.
Tools* e System Prompt: quando vira comportamento (e não só função)
Até agora, o que existe são funções. Para virar um Agent, é preciso empacotar capacidades como tools. Uma tool é um contrato: a descrição não serve apenas para documentar, ela orienta o modelo sobre quando e como usar cada recurso, quais entradas são esperadas e que tipo de saída deve ser produzida.
Quando você explicita dependências, por exemplo, que get_route e draft_samu_message devem ser chamadas depois de geo_analyze_impact, você reduz erros clássicos do fluxo, como rota sem destino definido, mensagem sem contexto e respostas que soam bem, mas não estão ancoradas nos dados.
TOOLS = [
{
"type": "function",
"name": "geo_analyze_impact",
"description": (
"Analisa impacto em vias (OSM) num buffer ao redor do acidente "
"e encontra o hospital mais próximo."
),
"parameters": {
"type": "object",
"properties": {
"lat": {"type": "number", "description": "Latitude WGS84."},
"lon": {"type": "number", "description": "Longitude WGS84."},
"buffer_m": {"type": "integer", "description": "Raio em metros (padrão 250)."},
"road_tags": {
"type": "array",
"items": {"type": "string",
"enum": ["motorway","trunk","primary","secondary",
"tertiary","residential","service","unclassified"]},
"description": "Classes OSM para filtrar. Padrão: todas acima."
}
},
"required": ["lat", "lon", "buffer_m"]
}
},
{
"type": "function",
"name": "get_route",
"description": (
"Calcula rota viária (OSRM) da origem ao hospital mais próximo, "
"considerando as vias disponíveis. Chame APÓS geo_analyze_impact."
),
"parameters": {
"type": "object",
"properties": {
"origin_lat": {"type": "number"},
"origin_lon": {"type": "number"},
"dest_lat": {"type": "number"},
"dest_lon": {"type": "number"},
},
"required": ["origin_lat", "origin_lon", "dest_lat", "dest_lon"]
}
},
{
"type": "function",
"name": "draft_samu_message",
"description": (
"Gera mensagem FICTÍCIA/EDUCACIONAL ao SAMU com base no resultado geo. "
"Chame APÓS geo_analyze_impact."
),
"parameters": {
"type": "object",
"properties": {
"geo_result": {"type": "object",
"description": "JSON retornado por geo_analyze_impact."}
},
"required": ["geo_result"]
}
},
]
Tools prontas, o proximo passo é definir o System Prompt que estabelece o comportamento em fluxos que espelham cenários reais na interface, no nosso caso:
- Testemunha, com triagem guiada;
- Emergência, com execução imediata (sem perguntas);
- Pedido simples de rota.
As regras gerais, como não responder vazio, encerrar com aviso, não inventar dados e reforçar que é uma simulação, trazem "disciplina" ao sistema e ajudam a manter consistência ao longo das interações.
SYSTEM = """
Você é um agente de emergência GIS (SIMULAÇÃO EDUCACIONAL).
Seu objetivo é auxiliar em situações de acidente de trânsito com base em dados de localização.
## Fluxo 1 — PRESENCIEI UM ACIDENTE
Quando o usuário informar que presenciou um acidente e fornecer respostas às perguntas de triagem
(consciência, sangramento, mobilidade, número de vítimas), você deve:
1. Chamar `geo_analyze_impact` para mapear as vias afetadas.
2. Chamar `draft_samu_message` com os dados de triagem relatados.
3. Responder com um resumo claro: estado das vítimas, hospital mais próximo e orientações
de primeiros socorros básicos até o SAMU chegar.
## Fluxo 2 — EMERGÊNCIA COM DADOS DA VÍTIMA
Quando receber uma mensagem iniciando com "EMERGÊNCIA — execute imediatamente as tools",
você DEVE obrigatoriamente chamar as tools na ordem indicada sem fazer perguntas:
1. `geo_analyze_impact` com lat/lon fornecidos.
2. `get_route` para o hospital mais próximo encontrado.
3. `draft_samu_message` com os dados da vítima.
Depois retorne um resumo completo: dados da vítima, hospital, distância, rota no mapa
e mensagem modelo para o SAMU.
## Fluxo 3 — VER ROTA
Quando solicitado, chamar `get_route` e informar o hospital destino e distância estimada.
## Regras gerais
- NUNCA retorne uma resposta vazia ou apenas "ok". Sempre retorne um resumo útil.
- Sempre encerre respostas de emergência com: "⚠️ Em emergência real, ligue 192 imediatamente."
- Seja objetivo e claro. Priorize informações de segurança.
- Não invente dados que não foram fornecidos.
- Este é um sistema de SIMULAÇÃO — nunca afirme que o SAMU foi realmente acionado.
- Ao resumir o resultado, SEMPRE cite os reference_points retornados pelo geo_analyze_impact
como pontos de referência para localizar a vítima (ex: "próximo à Farmácia X e ao Posto Y").
"""
O loop do agente: pensar, agir, observar e concluir
O comportamento do Agent depende de um loop que permite múltiplas etapas dentro de uma mesma solicitação. O modelo recebe o histórico e a lista de tools, decide quais chamadas fazer, você executa, devolve os resultados e ele continua iterando até produzir a resposta final.
Esse desenho traz previsibilidade, porque você controla o limite de rodadas e evita que um chat de alertas acabe virando um bate-papo sem fim. Ao mesmo tempo, mantém a flexibilidade necessária para encadear ações, já que uma etapa pode depender diretamente do resultado da anterior. Além disso, manter conversa e resultados como histórico ajuda a resposta a permanecer coerente entre interações, reduzindo contradições e evitando que o modelo perca o contexto no meio do fluxo.
def agent_turn(history, state, lat, lon, buf):
messages = [{"role": "system", "content": SYSTEM}, *history]
for step in range(6): # máximo de rodadas
resp = client.responses.create(
model="gpt-4.1-mini",
input=messages,
tools=TOOLS,
tool_choice="auto",
)
# Coleta tool calls dessa rodada
tool_calls = [
item for item in (getattr(resp, "output", []) or [])
if getattr(item, "type", None) == "function_call"
]
# Se não tem tool calls, tenta extrair texto e retorna
if not tool_calls:
text = extract_text(resp)
if text:
return text
# Fallback: modelo parou mas sem texto
return "⚠️ O agente finalizou sem gerar resposta textual."
# Adiciona output do modelo ao histórico
messages += list(getattr(resp, "output", []) or [])
# Executa cada tool call e adiciona resultado
for tc in tool_calls:
try:
args = tc.arguments
if isinstance(args, str):
args = json.loads(args)
result = run_tool(tc.name, args or {}, state, lat, lon, buf)
output = json.dumps(json_sanitize(result), ensure_ascii=False)
except Exception as ex:
output = f"Erro ao executar {tc.name}: {ex}"
messages.append({
"type": "function_call_output",
"call_id": tc.call_id,
"output": output,
})
# Após tools, verifica se já há texto na resposta atual
text = extract_text(resp)
if text:
return text
return "⚠️ O agente não finalizou após o limite de rodadas."
Também vale tratar um ponto prático: é preciso extrair o texto com segurança e evitar a sensação de travamento quando o SDK muda o formato de resposta, o que acontece na prática. Ao cobrir diferentes estruturas possíveis (texto direto, blocos, mensagens e recusas), a aplicação fica mais resiliente a variações do retorno da API.
def extract_text(resp) -> str:
# caminho mais simples SDK novo
txt = getattr(resp, "output_text", None)
if isinstance(txt, str) and txt.strip():
return txt.strip()
# varre output
out = []
for item in getattr(resp, "output", []) or []:
itype = getattr(item, "type", None)
# Alguns SDKs retornam output_text como item direto
if itype in ("output_text", "text"):
t = getattr(item, "text", None)
if isinstance(t, str) and t.strip():
out.append(t.strip())
continue
# Mensagem com blocos
if itype == "message":
for block in getattr(item, "content", []) or []:
btype = getattr(block, "type", None)
# blocos textuais comuns
if hasattr(block, "text") and isinstance(block.text, str) and block.text.strip():
out.append(block.text.strip())
continue
if btype in ("output_text", "text"):
t = getattr(block, "text", None)
if isinstance(t, str) and t.strip():
out.append(t.strip())
continue
# quando o modelo recusa por segurança (às vezes vem assim)
if btype == "refusal":
r = getattr(block, "refusal", None) or getattr(block, "text", None)
if isinstance(r, str) and r.strip():
out.append(r.strip())
continue
# Em alguns casos, recusa vem como item direto
if itype == "refusal":
r = getattr(item, "refusal", None) or getattr(item, "text", None)
if isinstance(r, str) and r.strip():
out.append(r.strip())
continue
return "\n".join(out).strip()
Memória de curto prazo: estado compartilhado para continuidade
Um Agent sem memória de curto prazo vira um sistema cansativo, porque o usuário acaba tendo de repetir o contexto o tempo todo. Por isso, faz sentido criar uma função que armazene estado e resultados recentes, permitindo chamadas dependentes sem exigir entradas “perfeitas” a cada etapa.
Além disso, o tool executor pode normalizar parâmetros e reaproveitar o que já foi descoberto no fluxo. Se a chamada de rota não trouxer um destino explícito, por exemplo, ele recupera o hospital salvo no estado. Se a mensagem precisar de contexto, ele combina o que o usuário enviou com o último resultado geográfico disponível. Esse reaproveitamento deixa o fluxo mais fluido e aproxima a experiência do que se espera em uma aplicação real, em que o sistema mantém continuidade e reduz atrito.
def run_tool(name, args, state, lat, lon, buf):
if name == "geo_analyze_impact":
args.setdefault("lat", lat); args.setdefault("lon", lon)
args.setdefault("buffer_m", buf)
valid = set(DEFAULT_ROAD_TAGS)
rt = [t for t in args.get("road_tags", []) if t in valid]
args["road_tags"] = rt if rt else DEFAULT_ROAD_TAGS
out, area, roads, hosps = compute_impact(
lat=args["lat"], lon=args["lon"],
buf=args["buffer_m"], road_tags=args["road_tags"]
)
state.update({"last_geo": out, "last_area": area,
"last_roads": roads, "last_hosps": hosps})
return out
if name == "get_route":
args.setdefault("origin_lat", lat); args.setdefault("origin_lon", lon)
if "dest_lat" not in args:
nh = state.get("last_geo", {}).get("nearest_hospital")
if not nh: raise ValueError("Sem hospital no state para get_route.")
args["dest_lat"] = nh["lat"]; args["dest_lon"] = nh["lon"]
out = get_route(args["origin_lat"], args["origin_lon"],
args["dest_lat"], args["dest_lon"])
state["last_route"] = out
return out
if name == "draft_samu_message":
geo_from_state = state.get("last_geo", {})
geo_result = {**geo_from_state, **args.get("geo_result", {})}
nh = geo_result.get("nearest_hospital")
if nh:
nome_hospital = nh.get("name") or "Hospital sem nome cadastrado"
ht = f"{nome_hospital} (~{nh['dist_m']:.0f}m)"
else:
ht = "nenhum hospital encontrado"
refs = geo_result.get("reference_points", [])
if refs:
ref_txt = "\nReferências próximas: " + " | ".join(
f'{r["name"]} ({r["type"]}, ~{r["dist_m"]:.0f}m)' for r in refs
)
else:
ref_txt = ""
acc = geo_result.get("accident", {})
return {"message": (
"EXEMPLO EDUCACIONAL (não real)\n\n"
f'SAMU 192: "Sofri um acidente. '
f'Localização: lat {acc.get("lat",0):.6f}, lon {acc.get("lon",0):.6f}. '
f'Hospital de referência: {ht}.{ref_txt} Aguardo socorro."\n\n'
"⚠️ Em emergência real, ligue 192 imediatamente."
)}
raise ValueError(f"Tool desconhecida: {name}")
Interface: a aplicação como camada final de operação
Na camada de interface, a prioridade não foi refinar engenharia de front-end, e sim entregar funcionalidade. Usei o ChatGPT como apoio para montar a UI, com mapa em camadas, histórico em formato de chat e botões de ação bem destacados para acionar os modos "Presenciei acidente”, "Tive um acidente" e "Ver rota".
O ponto central é que a UI cumpre o papel de painel operacional da simulação, serve exatamente para guiar a triagem, dispara execução imediata quando necessário e exibe resultados verificáveis no mapa. Isso ajuda a validar o que o agente inferiu e o que de fato foi calculado pelas tools.
Casos de uso
Para demonstrar o comportamento operacional do sistema, a simulação foi organizada em dois casos de uso que representam situações comuns em emergências. O objetivo não é apenas “mostrar o mapa”, mas evidenciar como o Agent muda sua postura conforme o contexto:
- Caso de uso 1: Testemunha com triagem guiada;
- Caso de uso 2: Vítima com execução imediata e rota no mapa
Eu sei que a tirinha ta exagerada, mas ilustra bem a ideia.
Caso de Uso 1: Testemunha com triagem guiada
O usuário entra como testemunha, responde perguntas de triagem e recebe um resumo com orientações. No mapa aparecem hospital e pontos de referência, tornando a localização mais compreensível. A mensagem modelo mostra como o resultado geográfico vira comunicação objetiva.
Caso de Uso 2: vítima com execução imediata e rota no mapa
O usuário aciona o modo emergência, por exemplo via controle de voz, e o Agent executa a sequência completa: análise do entorno, rota e mensagem modelo. A diferença é a ausência de perguntas, ele pegaria as informações do usuário direto do aparelho. O sistema age, desenha a rota e entrega um resumo com destino, distância, duração estimada e referências.
Conclusão
O valor deste projeto não está apenas em aplicar geoprocessamento, nem apenas em usar um modelo de linguagem. Ele está em transformar funções isoladas em um sistema com comportamento operacional. As tools definem o que o Agent consegue fazer, o prompt define como ele deve agir, o loop permite encadear decisões, o estado garante continuidade e a interface torna tudo utilizável e verificável no mapa.
Mesmo sendo uma simulação educacional, a proposta já indica um caminho claro para evoluir para uma aplicação mais madura. Ela pode se transformar em um produto real ao manter o mesmo núcleo, isto é, respostas baseadas em dados estruturados, comunicação objetiva, visualização espacial e mecanismos de resiliência para lidar com falhas de APIs e variações do mundo real. A partir daí, basta acrescentar camadas operacionais, como observabilidade, auditoria, testes, governança de mensagens e integrações.
