Content
# 🤖 Telecom AI Platform
> Plataforma de IA para telecomunicações combinando **FastAPI**, **LangGraph** e **MCP** (Model Context Protocol).
📚 **Projeto didático** — cada arquivo Python explica POR QUE cada decisão foi tomada, o que um Senior faria vs erros comuns de Junior.
---
## ⚡ Quick Start
```bash
# 1️⃣ Clone e entre no diretório
git clone https://github.com/Finish-Him/telecom-ai-platform.git
cd telecom-ai-platform
# 2️⃣ Crie ambiente virtual (recomendado)
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
# 3️⃣ Instale dependências
pip install -e ".[dev]"
# 4️⃣ Configure variáveis de ambiente
cp .env.example .env
# ✅ Modo mock funciona sem nenhuma mudança!
# 5️⃣ Rode o servidor
uvicorn app.main:app --reload
# 6️⃣ Abra no navegador
# 🌐 http://localhost:8000/docs → Swagger UI
# 💚 http://localhost:8000/health → Health check
```
---
## 🛠️ Stack Técnica
| Tecnologia | Propósito | Por quê? |
|:---:|---|---|
| 🚀 **FastAPI** | Framework web async | Clean Architecture + OpenAPI automático |
| 🔀 **LangGraph** | Workflow multi-agente | StateGraph: classify → route → process |
| 🔌 **FastMCP v3** | MCP Server | Expõe tools para modelos de IA |
| 🔍 **ChromaDB** | Vector database embedded | Busca semântica sem servidor externo |
| ✅ **Pydantic v2** | Validação de dados | Settings tipados + schemas |
| 📋 **structlog** | Logging estruturado | JSON logs para observabilidade |
| 🧪 **pytest** | Testes async | 9 testes com cobertura |
---
## 🏗️ Arquitetura
```
┌───────────────────┐
│ 🚀 FastAPI (8000) │
│ /api/v1/agents │
│ /api/v1/knowledge │
└────────┬──────────┘
│
┌────────▼──────────┐
│ 📦 Agent Service │
└────────┬──────────┘
│
┌────────▼──────────┐
│ 🔀 LangGraph │
│ StateGraph │
└───┬──────────┬────┘
│ │
┌────────▼──┐ ┌───▼────────┐
│ 🏷️ Classify│ │ ⚙️ Process │
│ Agent │ │ Agent │
└────────┬──┘ └───┬────────┘
│ │
│ ┌──────▼───────┐
│ │ 🔍 ChromaDB │
│ │ (Vector DB) │
│ └──────────────┘
│
┌────────▼──────────┐
│ 🤖 LLM Service │
│ (Mock Mode) │
└───────────────────┘
Separado:
┌───────────────────┐
│ 🔌 MCP Server │
│ (8001) │
│ • search_kb │
│ • classify_ticket │
└───────────────────┘
```
### 🔀 Fluxo do LangGraph
```
START → 🏷️ Classify → 🔀 Route by Priority
│
┌─────────┴─────────┐
│ │
🟡 Normal 🔴 Urgent
│ │
⚙️ Process ⚡ Process
Normal Urgent
│ │
└─────────┬─────────┘
│
END
```
---
## 📡 Endpoints da API
### 🔓 Públicos (sem auth)
| Método | Path | Descrição |
|:---:|---|---|
| 💚 GET | `/api/v1/health` | Health check + status das dependências |
| 🔍 POST | `/api/v1/knowledge/search` | Busca semântica na base de conhecimento |
| 📊 GET | `/api/v1/knowledge/stats` | Estatísticas da KB |
### 🔐 Protegidos (JWT)
| Método | Path | Descrição |
|:---:|---|---|
| 🔑 POST | `/api/v1/auth/token` | Obter JWT (demo: `admin` / `admin123`) |
| 🤖 POST | `/api/v1/agents/process-ticket` | Processa ticket pelo workflow LangGraph |
| 📥 POST | `/api/v1/knowledge/ingest` | Adiciona documento à base |
### 💡 Exemplo de uso
```bash
# 1. Obter token
TOKEN=$(curl -s -X POST http://localhost:8000/api/v1/auth/token \
-H "Content-Type: application/json" \
-d '{"username": "admin", "password": "admin123"}' | jq -r .access_token)
# 2. Processar ticket
curl -X POST http://localhost:8000/api/v1/agents/process-ticket \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"ticket_text": "Minha internet fibra está sem conexão desde ontem", "customer_id": "cli-001"}'
```
---
## 🔌 MCP Server
O MCP Server roda como **processo separado** na porta 8001:
```bash
python -m app.mcp.server
# ou: telecom-mcp (se instalado com pip install -e .)
```
### 🧰 Tools disponíveis para modelos AI
| Tool | Descrição |
|---|---|
| 🔍 `search_knowledge_base(query, limit)` | Busca semântica na KB |
| 🏷️ `classify_ticket(ticket_text)` | Classifica ticket (categoria + prioridade) |
| 📋 `get_ticket_categories()` | Lista categorias e prioridades disponíveis |
| 📊 `get_knowledge_base_stats()` | Estatísticas da base de conhecimento |
---
## 📁 Estrutura do Projeto
```
telecom-ai-platform/
├── 📂 app/
│ ├── main.py # 🚀 Entry point (lifespan, CORS, routers)
│ ├── config.py # ⚙️ Pydantic Settings + @lru_cache
│ ├── dependencies.py # 💉 Container DI (Depends)
│ ├── 📂 api/v1/
│ │ ├── router.py # 🔗 Agrega todos os routers v1
│ │ ├── 📂 endpoints/
│ │ │ ├── health.py # 💚 GET /health
│ │ │ ├── auth.py # 🔑 POST /auth/token
│ │ │ ├── agents.py # 🤖 POST /agents/process-ticket
│ │ │ └── knowledge.py # 📚 /knowledge/search, /ingest, /stats
│ │ └── 📂 schemas/
│ │ ├── requests.py # 📥 Pydantic request models
│ │ └── responses.py # 📤 Pydantic response models
│ ├── 📂 core/
│ │ ├── security.py # 🔐 JWT + OAuth2PasswordBearer
│ │ ├── middleware.py # 📋 Logging middleware + error handler
│ │ └── exceptions.py # ❌ Custom exception hierarchy
│ ├── 📂 services/
│ │ ├── agent_service.py # 🤖 Ponte HTTP ↔ LangGraph
│ │ ├── embedding_service.py # 🔍 Wrapper ChromaDB
│ │ └── llm_service.py # 🧠 Abstração LLM (mock + real)
│ ├── 📂 repositories/
│ │ ├── base.py # 🏛️ ABC Generic[T] repository
│ │ └── vector_repository.py # 🔍 ChromaDB PersistentClient
│ ├── 📂 agents/
│ │ ├── state.py # 📊 TicketState (TypedDict)
│ │ ├── classifier_agent.py # 🏷️ Nó: classifica ticket
│ │ ├── processor_agent.py # ⚙️ Nó: RAG (busca KB + resposta)
│ │ └── orchestrator.py # 🔀 StateGraph completo
│ └── 📂 mcp/
│ ├── server.py # 🔌 FastMCP("Telecom AI Tools")
│ └── 📂 tools/
│ ├── search_tool.py # 🔍 search_knowledge_base()
│ └── classify_tool.py # 🏷️ classify_ticket()
├── 📂 tests/ # 🧪 pytest async (9 testes)
├── 📂 data/
│ └── seed_knowledge.json # 📚 12 artigos telecom PT-BR
├── pyproject.toml # 📦 PEP 621 + ruff + pytest
├── .env.example # 🔧 Template de variáveis
└── README.md # 📖 Você está aqui!
```
---
## 🎓 Conceitos Demonstrados
| Conceito | Onde | O que aprende |
|---|---|---|
| 🏛️ **Clean Architecture** | `api/` → `services/` → `repositories/` | Camadas separadas com responsabilidades claras |
| 💉 **Dependency Injection** | `dependencies.py` | FastAPI Depends() para desacoplamento |
| 🗄️ **Repository Pattern** | `repositories/base.py` | ABC + Generic[T] para abstrair banco |
| 🔀 **LangGraph StateGraph** | `agents/orchestrator.py` | Workflow multi-agente com conditional routing |
| 🔌 **MCP Protocol** | `mcp/server.py` | Tools expostas para modelos de IA |
| 🔍 **RAG** | `agents/processor_agent.py` | Retrieval Augmented Generation com ChromaDB |
| 🔐 **JWT Auth** | `core/security.py` | Autenticação stateless com OAuth2 |
| 📋 **Structured Logging** | `core/middleware.py` | structlog com JSON |
| ⚙️ **12-Factor App** | `config.py` + `.env` | Configuração via ambiente |
| ✅ **Pydantic v2** | `schemas/` + `config.py` | Validação, schemas, settings |
| 🏭 **Closure Pattern** | `classifier_agent.py` | DI em nós do LangGraph |
| 🧪 **Async Testing** | `tests/` | pytest-asyncio + httpx AsyncClient |
---
## 🧪 Testes
```bash
# ▶️ Rodar testes
pytest -v
# 📊 Com cobertura
pytest --cov=app
# 🔍 Teste específico
pytest tests/test_agents.py -v
```
**Resultado esperado:** ✅ 9 testes passando (~4s)
```
tests/test_agents.py::test_mock_classify_internet ✅
tests/test_agents.py::test_mock_classify_financeiro ✅
tests/test_agents.py::test_mock_classify_cancelamento ✅
tests/test_agents.py::test_process_ticket_requires_auth ✅
tests/test_agents.py::test_process_ticket_success ✅
tests/test_agents.py::test_process_ticket_validation ✅
tests/test_agents.py::test_knowledge_search ✅
tests/test_health.py::test_health_returns_200 ✅
tests/test_health.py::test_health_has_dependencies ✅
```
---
## 🚀 Próximos Passos
- [ ] 🐳 Docker + docker-compose
- [ ] 🌐 Integração com LLM real (OpenRouter/Anthropic)
- [ ] 📊 Métricas com Prometheus
- [ ] 🔄 WebSocket para respostas em streaming
- [ ] 👤 Human-in-the-loop no LangGraph
- [ ] 📈 Dashboard de tickets processados
---
## 👨💻 Autor
**Moises Costa** — Senior Python Developer
[](https://github.com/Finish-Him)
