Xiaohongshu MCP Python

# 小红书运营 Agent 技术栈与项目结构 ## 技术栈选型 ### 核心框架 - **LangGraph** (>=0.2.0) - 工作流编排核心 - **LangChain** (>=0.3.0) - LLM 集成与工具链 - **Python** 3.11+ - 运行环境 ### AI 模型 - **OpenAI GPT-4** / **Anthropic Claude** - 内容生成与决策 - **本地嵌入模型** (sentence-transformers) - 语义检索 - **qwen-vl** / **cogview** - 图片生成 (可选) ### 数据存储 - **PostgreSQL** + **pgvector** - 向量数据库 (内容库/历史记录) - **Redis** - 状态缓存与消息队列 - **MinIO** / **OSS** - 媒体文件存储 ### 任务调度 - **Celery** + **Redis** - 异步任务队列 - **APScheduler** - 定时任务调度 ### API 框架 - **FastAPI** - Web API 服务 - **Pydantic** V2 - 数据验证与序列化 ### 监控与日志 - **Prometheus** + **Grafana** - 指标监控 - **Structlog** - 结构化日志 - **Sentry** - 错误追踪 ### 开发工具 - **Poetry** - 依赖管理 - **Ruff** - 代码格式化与检查 - **Pytest** - 单元测试 - **Pre-commit** - Git 钩子 --- ## 项目目录结构 (src 布局) ``` xiaohongshu-agent/ │ ├── pyproject.toml # Poetry 配置 ├── poetry.lock ├── README.md ├── .env.example # 环境变量模板 ├── .gitignore ├── docker-compose.yml # 本地开发环境 │ ├── src/ │ └── xiaohongshu_agent/ # 主包 │ │ │ ├── __init__.py │ ├── __main__.py # CLI 入口 │ │ │ ├── core/ # 核心功能 │ │ ├── __init__.py │ │ ├── config.py # 配置管理 (Pydantic Settings) │ │ ├── logging.py # 日志配置 │ │ ├── exceptions.py # 自定义异常 │ │ └── constants.py # 常量定义 │ │ │ ├── graph/ # LangGraph 工作流 │ │ ├── __init__.py │ │ ├── workflow.py # 主工作流定义 │ │ ├── state.py # 状态模型定义 │ │ ├── nodes/ # 节点实现 │ │ │ ├── __init__.py │ │ │ ├── entry.py # 入口路由节点 │ │ │ ├── supervisor.py # 协调节点 │ │ │ ├── content_agent.py # 内容 Agent │ │ │ ├── interaction_agent.py # 互动 Agent │ │ │ ├── analysis_agent.py # 分析 Agent │ │ │ └── state_manager.py # 状态管理节点 │ │ │ │ │ └── edges/ # 条件边逻辑 │ │ ├── __init__.py │ │ ├── routing.py # 路由决策 │ │ └── conditions.py # 条件判断 │ │ │ ├── agents/ # Agent 子状态机 │ │ ├── __init__.py │ │ ├── base.py # Agent 基类 │ │ │ │ │ ├── content/ # 内容 Agent │ │ │ ├── __init__.py │ │ │ ├── strategy.py # 策略规划 │ │ │ ├── topic.py # 话题选择 │ │ │ ├── generator.py # 内容生成 │ │ │ ├── material.py # 素材生成 │ │ │ ├── review.py # 内容审核 │ │ │ └── publisher.py # 发布执行 │ │ │ │ │ ├── interaction/ # 互动 Agent │ │ │ ├── __init__.py │ │ │ ├── analyzer.py # 内容分析 │ │ │ ├── responder.py # 响应策略 │ │ │ └── executor.py # 执行回复 │ │ │ │ │ └── analysis/ # 分析 Agent │ │ ├── __init__.py │ │ ├── collector.py # 数据收集 │ │ ├── processor.py # 数据处理 │ │ ├── insights.py # 模式识别 │ │ ├── optimizer.py # 决策建议 │ │ └── monitor.py # 异常检测 │ │ │ ├── tools/ # LangChain Tools │ │ ├── __init__.py │ │ ├── xiaohongshu_mcp.py # 小红书 MCP 工具 │ │ ├── web_search.py # 网络搜索 │ │ ├── image_gen.py # 图片生成 │ │ └── sentiment.py # 情感分析 │ │ │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ ├── content.py # 内容模型 │ │ ├── interaction.py # 互动模型 │ │ ├── analytics.py # 分析模型 │ │ └── user.py # 用户模型 │ │ │ ├── services/ # 业务服务层 │ │ ├── __init__.py │ │ ├── llm_service.py # LLM 调用服务 │ │ ├── storage_service.py # 存储服务 │ │ ├── cache_service.py # 缓存服务 │ │ └── notification_service.py # 通知服务 │ │ │ ├── repositories/ # 数据访问层 │ │ ├── __init__.py │ │ ├── base.py # 基础仓储 │ │ ├── content_repo.py # 内容仓储 │ │ ├── interaction_repo.py # 互动仓储 │ │ └── analytics_repo.py # 分析仓储 │ │ │ ├── tasks/ # Celery 任务 │ │ ├── __init__.py │ │ ├── celery_app.py # Celery 配置 │ │ ├── scheduled.py # 定时任务 │ │ └── background.py # 后台任务 │ │ │ ├── api/ # FastAPI 应用 │ │ ├── __init__.py │ │ ├── app.py # FastAPI 应用实例 │ │ ├── dependencies.py # 依赖注入 │ │ │ │ │ ├── routes/ # API 路由 │ │ │ ├── __init__.py │ │ │ ├── content.py # 内容管理 │ │ │ ├── interaction.py # 互动管理 │ │ │ ├── analytics.py # 数据分析 │ │ │ └── workflow.py # 工作流控制 │ │ │ │ │ └── schemas/ # API Schema │ │ ├── __init__.py │ │ ├── requests.py # 请求模型 │ │ └── responses.py # 响应模型 │ │ │ └── utils/ # 工具函数 │ ├── __init__.py │ ├── text.py # 文本处理 │ ├── image.py # 图片处理 │ ├── retry.py # 重试逻辑 │ └── validators.py # 验证器 │ ├── tests/ # 测试目录 │ ├── __init__.py │ ├── conftest.py # Pytest 配置 │ │ │ ├── unit/ # 单元测试 │ │ ├── test_agents/ │ │ ├── test_tools/ │ │ └── test_services/ │ │ │ ├── integration/ # 集成测试 │ │ ├── test_workflow.py │ │ └── test_api.py │ │ │ └── fixtures/ # 测试数据 │ └── mock_data.json │ ├── scripts/ # 脚本工具 │ ├── setup_db.py # 数据库初始化 │ ├── seed_data.py # 种子数据 │ └── deploy.sh # 部署脚本 │ ├── config/ # 配置文件 │ ├── development.yaml │ ├── production.yaml │ └── prompts/ # Prompt 模板 │ ├── content_generation.txt │ ├── comment_response.txt │ └── analysis_report.txt │ ├── docs/ # 文档 │ ├── architecture.md │ ├── api.md │ └── deployment.md │ └── docker/ # Docker 配置 ├── Dockerfile ├── Dockerfile.worker # Celery Worker └── nginx.conf ``` --- ## 核心依赖 (pyproject.toml 示例) ```toml [tool.poetry] name = "xiaohongshu-agent" version = "0.1.0" description = "基于 LangGraph 的小红书运营 Agent" authors = ["Your Name <you@example.com>"] packages = [{include = "xiaohongshu_agent", from = "src"}] [tool.poetry.dependencies] python = "^3.11" langgraph = "^0.2.0" langchain = "^0.3.0" langchain-openai = "^0.2.0" langchain-anthropic = "^0.2.0" fastapi = "^0.115.0" uvicorn = {extras = ["standard"], version = "^0.31.0"} pydantic = "^2.9.0" pydantic-settings = "^2.5.0" sqlalchemy = "^2.0.0" asyncpg = "^0.29.0" pgvector = "^0.3.0" redis = "^5.1.0" celery = {extras = ["redis"], version = "^5.4.0"} apscheduler = "^3.10.0" structlog = "^24.4.0" httpx = "^0.27.0" tenacity = "^9.0.0" pillow = "^11.0.0" [tool.poetry.group.dev.dependencies] pytest = "^8.3.0" pytest-asyncio = "^0.24.0" pytest-cov = "^6.0.0" ruff = "^0.7.0" pre-commit = "^4.0.0" ipython = "^8.28.0" [tool.ruff] line-length = 100 target-version = "py311" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" ``` --- ## 快速启动命令 ```bash # 安装依赖 poetry install # 启动本地开发环境 docker-compose up -d # 运行 FastAPI 服务 poetry run uvicorn src.xiaohongshu_agent.api.app:app --reload # 启动 Celery Worker poetry run celery -A src.xiaohongshu_agent.tasks.celery_app worker -l info # 启动 Celery Beat (定时任务) poetry run celery -A src.xiaohongshu_agent.tasks.celery_app beat -l info # 运行测试 poetry run pytest # 代码检查 poetry run ruff check src/ ``` --- ## 关键设计说明 ### 1. **src 布局优势** - 避免测试时意外导入源码目录 - 强制正确的包安装流程 - 更清晰的项目结构 ### 2. **分层架构** - **Graph 层**: 工作流编排逻辑 - **Agent 层**: 具体业务实现 - **Service 层**: 基础服务抽象 - **Repository 层**: 数据访问 ### 3. **配置管理** - 使用 Pydantic Settings 管理配置 - 支持环境变量覆盖 - 多环境配置文件 ### 4. **异步优先** - FastAPI 原生支持异步 - SQLAlchemy 2.0 异步模式 - LangChain 异步调用 ### 5. **可测试性** - 依赖注入设计 - Mock 友好的接口 - 完整的测试覆盖 这个结构既保证了代码的清晰性,又具备良好的扩展性和可维护性。