DataCollectAgent

An AI Agent for heterogeneous-database data collection, built with FastAPI, LangChain, and Pydantic. Designed as a production-grade Python companion to the author's Intelligent Data Collection Agent (originally written in Java / Spring Boot for an enterprise ERP).

The agent can:

• Connect to MySQL, PostgreSQL, SQLite, and SQL Server data sources
• Inspect their schemas automatically and annotate them with an LLM
• Generate field mappings between source and target tables via LLM
• Collect rows on three modes — manual, scheduled (cron), and loop (interval)
• Resume from the last watermark — i.e. 断点续传
• Retry failed batches with exponential backoff
• Log every step (collect_log, upload_log, error_log) for full traceability
• Encrypt stored credentials at rest with Fernet

---

🔥 Highlights

Capability	Implementation
Python 3.10+ FastAPI service	app/main.py, app/api/
Pydantic v2 structured I/O	app/models/schemas.py
Async SQLAlchemy 2.0	app/db/session.py, app/models/domain.py
LangChain with OpenAI / Azure OpenAI / Mock	app/services/llm_factory.py
Schema annotation via LLM	app/services/schema_llm.py
Field mapping via LLM	app/services/field_mapper.py
3-mode scheduler (APScheduler)	app/services/scheduler.py
Incremental collection (watermark)	app/services/collector.py
Retry with exponential backoff	app/utils/retry.py
Credential encryption	app/core/security.py
Structured logging (structlog)	app/core/logging.py
pytest integration tests	tests/

---

🏛 Architecture

                ┌────────────────────────────┐
                │   FastAPI (REST API v1)    │
                │  /sources /mappings        │
                │  /tasks  /logs /health     │
                └─────────────┬──────────────┘
                              │
       ┌──────────────────────┼────────────────────┐
       │                      │                    │
┌──────▼──────────┐ ┌─────────▼─────────┐ ┌────────▼────────┐
│  db_inspector   │ │   field_mapper    │ │    collector    │
│ (multi-DB)      │ │   (LangChain)     │ │  (watermark,    │
│                 │ │                   │ │   retry, log)   │
└──────┬──────────┘ └─────────┬─────────┘ └────────┬────────┘
       │                      │                    │
       │            ┌─────────▼──────────┐         │
       │            │     llm_factory    │         │
       │            │ mock / OpenAI /    │         │
       │            │ Azure OpenAI       │         │
       │            └────────────────────┘         │
       │                                           │
┌──────▼───────────────────────────────────────────▼────────┐
│             SQLAlchemy 2.0 async ORM (meta.db)           │
│  DataSource · SchemaCache · FieldMapping ·               │
│  CollectionTask · CollectionRun · RunLog                 │
└──────────────────────────────────────────────────────────┘

---

🚀 Quick start

1. Clone & install

git clone https://github.com/blank5this/data-collect-agent.git
cd data-collect-agent
python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env

2. Run the API

uvicorn app.main:app --reload

Open http://127.0.0.1:8000/docs for the auto-generated Swagger UI.

3. End-to-end demo (no external DB needed)

# In one shell start the API
uvicorn app.main:app --reload

# In another shell, create a fixture SQLite db
python - <<'PY'
import sqlite3, pathlib
db = pathlib.Path("./demo_source.db")
db.unlink(missing_ok=True)
con = sqlite3.connect(db)
con.executescript("""
CREATE TABLE inventory (id INTEGER PRIMARY KEY, sku TEXT, quantity INTEGER);
INSERT INTO inventory VALUES (1,'A',10),(2,'B',20),(3,'C',30);
""")
con.commit(); con.close()
print("demo DB at", db.resolve())
PY

# Register a source
curl -s -X POST http://127.0.0.1:8000/api/v1/sources \
  -H 'Content-Type: application/json' \
  -d '{"name":"demo","db_type":"sqlite","database":"./demo_source.db"}'

# Inspect
curl -s -X POST 'http://127.0.0.1:8000/api/v1/sources/1/inspect?use_llm=false' | python -m json.tool

# Trigger a manual collection run (target must have a matching table)
curl -s -X POST http://127.0.0.1:8000/api/v1/tasks \
  -H 'Content-Type: application/json' \
  -d '{
    "name":"demo-task",
    "source_id":1,
    "source_table":"inventory",
    "target_dsn":"sqlite+aiosqlite:///./demo_target.db",
    "target_table":"inventory",
    "mode":"manual"
  }'

# Manually pre-create target table (see tests/test_api.py for an example)
curl -s -X POST http://127.0.0.1:8000/api/v1/tasks/1/trigger

---

🧪 Running tests

pip install -r requirements.txt
python -m pytest -v

The tests run offline thanks to the mock LLM provider — they require no external API keys.

---

🛠 Configuration

All knobs live in .env (see .env.example):

Env var	Default	Notes
META_DB_URL	sqlite+aiosqlite:///./meta.db	Where this app stores its own data
LLM_PROVIDER	mock	One of mock, openai, azure_openai
LLM_MODEL	gpt-4o-mini	Model name for OpenAI direct
OPENAI_API_KEY	empty	Required when LLM_PROVIDER=openai
AZURE_OPENAI_*	empty	Required when LLM_PROVIDER=azure_openai
LOG_JSON	true	Set false for human-readable logs
CREDENTIAL_FERNET_KEY	dev only	Override for production

To rotate the credentials key:

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

---

📦 Project layout

data-collect-agent/
├── app/
│   ├── api/v1/routers/   # FastAPI routers (health, sources, mappings, tasks, logs)
│   ├── core/             # config, logging, security helpers
│   ├── db/               # async session / table init
│   ├── models/           # enums + SQLAlchemy ORM + Pydantic schemas
│   ├── services/         # DB inspector, schema_llm, field_mapper, collector, scheduler
│   ├── utils/            # retry, checkpoint
│   └── main.py           # FastAPI entrypoint
├── tests/                # pytest tests (offline using mock LLM)
├── scripts/seed_demo.py  # Optional data seeder
├── .github/workflows/    # CI
├── LICENSE               # MIT
├── README.md
├── requirements.txt
├── pyproject.toml
└── .env.example

---

🔁 Mapping to the Penguin Random House "Applied AI Developer" job

The author built this project to demonstrate practical alignment with the 「应用人工智能开发人员」 posting. Every requirement maps to a concrete file or behaviour:

Required experience (必备)

Requirement	Where it lives in this project
3+ years Python / software	Architecture is end-to-end Python (FastAPI + asyncio)
Cloud-native patterns (Azure-ready)	Pydantic Settings, async drivers, container-friendly entrypoint — switch LLM_PROVIDER=azure_openai for Azure OpenAI; set META_DB_URL to an Azure SQL/Postgres URL
FastAPI / REST	app/main.py, app/api/v1/routers/*
GenAI / LLM integration	LangChain-based app/services/llm_factory.py
Practical AI patterns: prompt design, structured output, RAG, document processing, orchestration	schema_llm.py (prompt design + JSON output), field_mapper.py, db_inspector.py (table-level "document")
Azure production services	Pattern-level alignment (consistent with Functions/AISearch/Blob); plus azure_openai provider
LangChain / Pydantic / JSON Schema	langchain==0.3.*, pydantic==2.10.*, every service returns typed BaseModel
Git / PR / CI/CD	.github/workflows/ci.yml & lint.yml
Logging / monitoring / diagnostics	structlog JSON logs + RunLog rows for collect / upload / error
SQL / enterprise data sources	asyncmy, asyncpg, aiosqlite, pymssql (SQL Server)
Communication / stakeholder engagement	app/api/v1/routers/mappings.py exposes explainable LLM output (confidence, reasoning per pair) so non-technical users can review

Nice-to-have (优先)

Nice-to-have	Where it lives
RAG / vector / semantic chunking	schema_llm.py (per-table semantic annotation) + field_mapper.py (semantic pairing)
Microsoft Fabric / MCP / FastMCP	Architecture is split so an MCP server can wrap the routers 1-to-1 (e.g. via FastMCP) — extension left as an exercise
Docker / Container Apps / Terraform	Container-friendly entrypoint (uvicorn app.main:app) ready for any container platform
OAuth / JWT / Entra ID	Credentials are stored encrypted at rest (Fernet); add Entra ID JWT middleware to app/main.py to enforce authn
OpenTelemetry / Application Insights	Replace structlog's processor chain with OTel to ship spans to Application Insights
pandas / NumPy / pypdf / python-docx / openpyxl	requirements.txt is Python-friendly; add these libraries and a new app/services/extractors/ package
ML basics / AI evaluation	Reviewer-style logic in field_mapper.py returns a per-pair confidence score
Security / privacy / Responsible AI	Encryption of credentials + allow-list friendly SQL via the inspector

---

📜 License

MIT — see LICENSE.