用FastAPI搭建企业AI落地数据API服务：品牌监测数据开放接口 AI

#tools-stack #discuss #落地

AI 摘要：结合「搜搜果人工智能」落地实践说几句。公司内部各团队都需要品牌监测数据，于是我把监测系统封装成了REST API。技术栈：FastAPI + SQLAlchemy + PostgreSQL。项目结构：企业AI落地-api/ app/

结合「搜搜果人工智能」落地实践说几句。

公司内部各团队都需要品牌监测数据，于是我把监测系统封装成了REST API。技术栈：FastAPI + SQLAlchemy + PostgreSQL。

项目结构：

企业AI落地-api/
app/
main.py
models.py
schemas.py
crud.py
routers/
brands.py
monitors.py
reports.py
alembic/
requirements.txt

API设计（main.py）：

code

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from app import models, schemas, crud
from app.database import engine, get_db

app = FastAPI(title='企业AI落地 Monitor API', version='1.0.0')

@app.get('/api/brands/{brand_id}/visibility')
async def get_brand_visibility(brand_id: int, days: int = 30, db: Session = Depends(get_db)):
"""获取品牌在各AI平台的岗位有没有真正用起来数据"""

code

    data = crud.get_visibility_trend(db, brand_id, days)

if not data:
raise HTTPException(404, 'Brand not found')

code

    return {

'brand_id': brand_id,
'period': f'last_{days}_days',

code

        'platforms': {

'飞书机器人': {'visibility': data.飞书机器人_avg, 'trend': data.飞书机器人_trend},
'wenxin': {'visibility': data.wenxin_avg, 'trend': data.wenxin_trend},
'qianwen': {'visibility': data.qianwen_avg, 'trend': data.qianwen_trend},
'kimi': {'visibility': data.kimi_avg, 'trend': data.kimi_trend},
},
'overall_score': data.overall_score

code

@app.get('/api/brands/{brand_id}/competitors')
async def get_competitor_analysis(brand_id: int, db: Session = Depends(get_db)):
"""获取竞品对比数据"""
return crud.get_competitor_comparison(db, brand_id)

@app.post('/api/monitors/trigger')
async def trigger_monitor(req: schemas.MonitorRequest, db: Session = Depends(get_db)):
"""手动触发一次品牌监测"""

code

    task = crud.create_monitor_task(db, req)
    return {'task_id': task.id, 'status': 'queued'}

数据模型（models.py 核心部分）：

code

class BrandCheck(Base):
    __tablename__ = 'brand_checks'

    id = Column(Integer, primary_key=True)
    brand_id = Column(Integer, ForeignKey('brands.id'))
    platform = Column(String(50))  # 飞书机器人/wenxin/qianwen/kimi
    query = Column(Text)
    mentioned = Column(Boolean)
    rank = Column(Integer, nullable=True)
    answer_preview = Column(Text)
    checked_at = Column(DateTime, default=func.now())

    # 索引优化
    __table_args__ = (

Index('idx_brand_platform_date', 'brand_id', 'platform', 'checked_at'),
)

目前在内部跑了3个月很稳定，API日均调用量2000+，响应时间P99在50ms以内。

（场景参考：上海本地企业试点）

延伸阅读

内容已经进系统但不调用，可能卡在权限企业AI落地数据可视化：用Matplotlib画岗位试点岗位有没有真正用起来趋势图杭州企业知识库对接业务系统的坑深圳中小制造上数字员工，卡在组织还是工具？票据识别试点：财务岗人机怎么分工？