# m-agent

**Repository Path**: mcbstudy/m-agent

## Basic Information

- **Project Name**: m-agent
- **Description**: python的agent回答系统，接入客户端界面
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-15
- **Last Updated**: 2026-07-31

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# M-Agent

一个基于 FastAPI + LangChain + DeepSeek API 的最小对话接口案例，后续可以继续扩展成 RAG、工具调用、多 Agent 编排或企业内部智能体服务。

## 技术栈

- Python 3.13+
- uv：Python 项目和依赖管理
- FastAPI：HTTP API 服务
- LangChain：大模型调用抽象层
- langchain-deepseek：LangChain 官方 DeepSeek 集成包
- DeepSeek API：实际大模型服务
- redis-py：Redis 异步客户端，用于跨进程持久化聊天会话和消息

## 依赖说明

核心依赖已经写入 `pyproject.toml`：

```toml
dependencies = [
    "fastapi>=0.138.1",
    "fastapi-cli>=0.0.7",
    "langchain>=1.3.11",
    "langchain-deepseek>=0.1.0",
    "redis[hiredis]>=5.2.0",
    "pydantic-settings>=2.0.0",
    "uvicorn[standard]>=0.30.0",
]
```

开发测试依赖：

```toml
[dependency-groups]
dev = [
    "httpx>=0.27.0",
    "pytest>=8.0.0",
]
```

安装或同步依赖：

```bash
uv sync
```

## 环境变量

复制示例文件：

```bash
cp .env.example .env
```

然后填写 DeepSeek API Key：

```bash
DEEPSEEK_API_KEY=sk-your-deepseek-api-key
DEEPSEEK_MODEL=deepseek-v4-flash
```

Redis 配置使用环境变量注入。你本地的 Docker Redis 可配置为：

```env
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_USERNAME=mcb
REDIS_PASSWORD=redis123456
REDIS_DB=0
REDIS_KEY_PREFIX=m-agent
# 本地开发默认无需请求头；生产环境必须改为 trusted_gateway。
CHAT_SCOPE_MODE=development
CHAT_DEFAULT_TENANT_ID=local
CHAT_DEFAULT_USER_ID=local-user
```

Redis 不可用时接口返回 503，不会自动回退到进程内存，避免多实例部署产生数据分裂。
`/health` 是进程存活检查，`/health/ready` 会额外检查 Redis 连通性；`/ready` 用于单独检查 MySQL 连通性。

常用配置：

| 变量名 | 说明 | 默认值 |
| --- | --- | --- |
| `DEEPSEEK_API_KEY` | DeepSeek API Key，必填 | 无 |
| `DEEPSEEK_MODEL` | DeepSeek 模型名 | `deepseek-v4-flash` |
| `DEEPSEEK_BASE_URL` | 自定义 API Base URL，通常可不填 | 无 |
| `DEEPSEEK_FAST_MODEL` | Fast 质量等级对应模型 | 无，回退到默认模型 |
| `DEEPSEEK_STRONG_MODEL` | Strong 质量等级对应模型 | 无，回退到默认模型 |
| `DEEPSEEK_TEMPERATURE` | 生成随机性，越低越稳定 | `0.2` |
| `DEEPSEEK_TIMEOUT_SECONDS` | 请求超时时间 | `30` |
| `DEEPSEEK_MAX_RETRIES` | 应用层模型重试次数；Provider SDK 不重复重试 | `2` |
| `DEEPSEEK_FALLBACK_MODEL` | 主模型失败后的备用模型 | 无 |
| `LLM_RATE_LIMIT_PER_SECOND` | 单模型每秒请求数限制 | `20` |
| `LLM_CIRCUIT_FAILURE_THRESHOLD` | 单模型连续失败多少次后熔断 | `5` |
| `LLM_CIRCUIT_OPEN_SECONDS` | 熔断冷却时间，单位秒 | `30` |
| `PROMPT_APPLICATION_NAME` | 服务端受控提示词中的应用名称 | `M-Agent` |
| `PROMPT_RESPONSE_LANGUAGE` | 服务端要求模型使用的默认回答语言 | `中文` |
| `PROMPT_MAX_HISTORY_MESSAGES` | 单次调用最多携带的历史消息条数 | `20` |
| `MYSQL_HOST` | MySQL 主机；同 Docker 网络中填服务名 | `127.0.0.1` |
| `MYSQL_PORT` | MySQL 端口 | `3306` |
| `MYSQL_DATABASE` | 默认连接数据库 | `opencode_db` |
| `MYSQL_USERNAME` | MySQL 业务账号 | `mcb` |
| `MYSQL_PASSWORD` | MySQL 业务账号密码，必填 | 无 |
| `MYSQL_CONNECT_TIMEOUT_SECONDS` | 建立连接超时（秒） | `5` |
| `MYSQL_POOL_SIZE` | 每个应用进程的常驻连接数 | `5` |
| `MYSQL_MAX_OVERFLOW` | 连接池繁忙时的临时连接上限 | `10` |
| `MYSQL_POOL_RECYCLE_SECONDS` | 连接复用最大时间（秒） | `1800` |
| `LANGSMITH_TRACING` | 是否启用 LangSmith Trace；必须写为小写 `true` | `false` |
| `LANGSMITH_API_KEY` | LangSmith API Key，启用 Trace 时必填 | 无 |
| `LANGSMITH_PROJECT` | Trace 归属项目；建议按环境区分 | `default` |
| `LANGSMITH_ENDPOINT` | 非美国区域或私有化部署的 LangSmith 地址 | 美国区域默认地址 |
| `LANGSMITH_WORKSPACE_ID` | API Key 关联多个 Workspace 时必填 | 无 |
| `REDIS_HOST` | Redis 地址 | `127.0.0.1` |
| `REDIS_PORT` | Redis 端口 | `6379` |
| `REDIS_USERNAME` | Redis ACL 用户名 | 无 |
| `REDIS_PASSWORD` | Redis ACL 密码 | 无 |
| `REDIS_DB` | Redis 逻辑数据库 | `0` |
| `REDIS_KEY_PREFIX` | Key 命名空间 | `m-agent` |
| `REDIS_MAX_CONNECTIONS` | 连接池上限 | `50` |
| `REDIS_SOCKET_TIMEOUT_SECONDS` | Redis 读写超时 | `2` |
| `REDIS_CONNECT_TIMEOUT_SECONDS` | Redis 连接超时 | `2` |
| `CHAT_SCOPE_MODE` | 身份范围来源：`development` 或 `trusted_gateway` | `development` |
| `CHAT_DEFAULT_TENANT_ID` | 开发模式未携带身份 Header 时的默认租户 | `local` |
| `CHAT_DEFAULT_USER_ID` | 开发模式未携带身份 Header 时的默认用户 | `local-user` |

`main.py` 会在本地启动时加载项目根目录下的 `.env`，让 LangSmith SDK 能读取到
`LANGSMITH_*`。加载不会覆盖 Docker、CI 或操作系统已经注入的同名环境变量，因此生产环境
仍应通过部署平台的 Secret 注入密钥。

说明：DeepSeek 官方文档目前推荐新的模型名，例如 `deepseek-v4-flash`。如果你使用旧文档或旧示例中常见的 `deepseek-chat`，需要注意官方废弃时间说明。

## 启动项目

开发模式启动：

```bash
uv run fastapi dev main.py
```

生产或本地稳定运行可以使用：

```bash
uv run uvicorn main:app --host 127.0.0.1 --port 8090
```

启动后访问：

- API 文档：http://127.0.0.1:8090/docs
- 健康检查：http://127.0.0.1:8090/health
- MySQL 就绪检查：http://127.0.0.1:8090/ready

`/health` 只验证应用进程本身是否存活；`/health/ready` 验证 Redis（聊天业务依赖），`/ready` 会通过连接池执行 `SELECT 1` 验证 MySQL 的网络、认证和数据库选择是否可用。当前仅完成 MySQL 基础设施接入，聊天会话和消息仍由 Redis 存储；未创建 MySQL 业务表，也未接入任何业务读写。

## Docker 部署环境变量

不要把 `.env`、DeepSeek Key 或 LangSmith Key 打进镜像或提交到仓库。生产环境应由
Docker Secret、Kubernetes Secret 或部署平台的环境变量功能注入。最小配置文件可放在服务器
受权限保护的位置，例如 `/opt/m-agent/.env.production`：

```env
DEEPSEEK_API_KEY=sk-***
DEEPSEEK_MODEL=deepseek-v4-flash
CHAT_SCOPE_MODE=trusted_gateway
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=lsv2-***
LANGSMITH_PROJECT=m-agent-prod
# 非美国区域才需要：
# LANGSMITH_ENDPOINT=https://eu.api.smith.langchain.com
```

使用现有镜像启动时，通过 `--env-file` 注入：

```bash
docker run -d \
  --name m-agent \
  --restart unless-stopped \
  --env-file /opt/m-agent/.env.production \
  -p 8090:8090 \
  your-registry/m-agent:latest
```

Docker Compose 的等价配置如下。生产环境将 `env_file` 指向宿主机的受保护文件；不要使用
仓库内的 `.env` 作为生产密钥文件。

```yaml
services:
  m-agent:
    image: your-registry/m-agent:latest
    ports:
      - "8090:8090"
    env_file:
      - /opt/m-agent/.env.production
    environment:
      LANGSMITH_TRACING: "true"
      LANGSMITH_PROJECT: m-agent-prod
    restart: unless-stopped
```

## 调用客户端对话接口

M-Chat 客户端发送消息时，会按下面的顺序调用服务端接口。

1. 创建会话：

```bash
curl -X POST "http://127.0.0.1:8090/api/v1/chat/sessions" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "LangChain 学习",
    "chatMode": "normal"
  }'
```

2. 保存用户消息：

```bash
curl -X POST "http://127.0.0.1:8090/api/v1/chat/messages" \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "上一步返回的 session id",
    "content": "请用 Java 后端工程师容易理解的方式解释 LangChain 的作用。",
    "chatMode": "normal"
  }'
```

3. 流式生成 assistant 回复：

```bash
curl -N -X POST "http://127.0.0.1:8090/api/v1/chat/stream" \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "上一步返回的 session id",
    "prompt": "请用 Java 后端工程师容易理解的方式解释 LangChain 的作用。",
    "chatMode": "normal"
  }'
```

响应事件示例：

```text
event: message_start
data: {"messageId": "msg-xxx", "sessionId": "session-xxx", "role": "assistant"}

event: token
data: {"messageId": "msg-xxx", "sessionId": "session-xxx", "content": "LangChain"}

event: message_end
data: {"messageId": "msg-xxx", "sessionId": "session-xxx", "finishReason": "stop", "model": "deepseek-v4-flash"}
```

事件约定：

- `message_start`：assistant 消息开始生成。
- `token`：模型生成的增量文本片段。
- `message_end`：模型输出完成，并携带本次使用的模型名。
- `error`：流式调用过程中发生异常，返回通用错误信息。

## 项目结构

```text
.
├── app
│   ├── api
│   │   ├── dependencies.py # FastAPI 基础设施依赖适配
│   │   └── http_router.py  # FastAPI HTTP 路由、SSE 转换和兼容入口
│   ├── common
│   │   ├── clock.py       # UTC 时间生成
│   │   ├── mysql.py       # MySQL 连接池、事务与生命周期
│   │   ├── observability.py # 日志与请求观测初始化
│   │   └── redis.py       # Redis 客户端、连接池与生命周期
│   ├── llm
│   │   ├── models.py      # 模型配置、请求和内部流式事件
│   │   ├── registry.py    # init_chat_model 初始化和模型缓存
│   │   ├── candidate_selection.py # 质量等级到模型候选链的选择策略
│   │   ├── failure_classification.py # Provider 失败分类与重试/降级决策
│   │   └── reliability.py # 重试、限流、熔断、降级和指标
│   ├── prompts
│   │   ├── templates.py   # 后端受控的 LangChain 提示词模板
│   │   ├── models.py      # PromptContext 等提示词领域对象
│   │   └── assembler.py   # 变量填充、历史边界控制和消息组装
│   ├── errors
│   │   ├── base.py        # ApplicationError 公共异常契约
│   │   ├── codes.py       # 稳定错误码与 HTTP 响应定义
│   │   ├── chat.py        # 聊天领域异常
│   │   ├── model.py       # 模型调用领域异常
│   │   ├── security.py    # 认证与访问作用域异常
│   │   ├── infrastructure.py # Redis、MySQL 等基础设施异常
│   │   └── handlers.py    # FastAPI 全局异常处理器
│   ├── config.py      # 环境变量配置
│   ├── chat_scope.py  # 租户/用户数据访问范围和身份上下文适配
│   ├── schemas.py     # 请求/响应模型
│   ├── chat_store.py  # ChatStore 存储端口和测试用内存替身
│   ├── redis_chat_store.py  # Redis Repository 适配器和 Key 设计
│   └── chat_generation.py # ChatGenerationService 应用服务门面
├── tests
│   └── test_api.py    # FastAPI 接口测试
├── main.py            # 应用启动入口和 FastAPI 装配
├── pyproject.toml     # 项目依赖声明
└── uv.lock            # 依赖锁文件
```

### Redis 数据结构

每个租户、用户使用独立的 Key 空间，当前用户的会话不会进入其他用户的 Sorted Set：

- `m-agent:chat:{tenant:acme:user:alice}:session:{sessionId}`：Hash，保存会话元数据。
- `m-agent:chat:{tenant:acme:user:alice}:session:{sessionId}:messages`：List，按写入顺序保存消息 JSON。
- `m-agent:chat:{tenant:acme:user:alice}:sessions:index`：Sorted Set，以更新时间作为 score 支持当前用户的倒序分页。

Key 中花括号内是 Redis Cluster Hash Tag。同一 `tenantId + userId` 的 Hash、List 和
Sorted Set 会落在相同 Hash Slot，因此包含多个 Key 的事务 Pipeline 和 Lua Script 都可直接
运行在 Redis Cluster 中。创建会话、追加消息使用事务 Pipeline，避免只写入部分结构。

### 会话更新性能与并发优化

`PATCH /api/v1/chat/sessions/{sessionId}` 只会修改 `title`、`saved` 和 `updatedAt`，响应也只
返回会话摘要，不再携带 `messages`。因此一次重命名或收藏不会读取整个消息 List，避免聊天历史
增长后出现无意义的 `LRANGE 0 -1` 开销。

更新操作通过 Redis Lua Script 原子执行：先判断当前 scope 下的会话 Hash 是否存在，再更新
Hash、刷新 Sorted Set score，并使用 `HMGET` 返回摘要。脚本执行期间 Redis 不会插入其他命令，
所以不会发生“检查存在后、写入前被另一请求删除，HSET 又重建残缺 Hash”的竞态问题。

```text
旧路径：HGETALL + LRANGE → HSET + ZADD → HGETALL + LRANGE
新路径：Lua（EXISTS → HSET → ZADD → HMGET）
```

前端收到摘要后只合并 `title`、`saved`、`updatedAt`，保留本地已加载的消息历史。

### 多租户数据隔离与认证边界

聊天接口会先解析 `ChatScope(tenantId, userId)`，再将该作用域传给 `ChatStore`。仓储层使用
scope 构造全部 Redis Key，因此即使调用方持有其他用户的 `sessionId`，也只能在自己的 Key
空间内查询，最终得到 404。

```text
JWT/OIDC 鉴权网关
  → 注入 X-Tenant-Id、X-User-Id
  → FastAPI get_chat_scope
  → ChatStore(scope, ...)
  → RedisKeyspace(scope)
  → 当前租户/用户专属 Redis Key
```

本地开发使用 `CHAT_SCOPE_MODE=development`，请求头缺失时会回退到默认本地身份；也可通过
`X-Tenant-Id`、`X-User-Id` 模拟多用户隔离。生产环境必须配置
`CHAT_SCOPE_MODE=trusted_gateway`，并让 API Gateway 在完成 JWT/OIDC 校验后覆盖这两个
Header，同时剥离外部客户端伪造的同名 Header。服务本身不验证 Header 签名，因此不能直接
暴露到公网后信任客户端传来的 Header。

旧版本的 `m-agent:chat:sessions:index` 是全局索引，缺少归属信息。升级后旧会话不会被新 Key
规则读取；生产迁移必须先从可信业务库补齐 `sessionId -> tenantId/userId` 映射，再写入新 Key。
无法确认归属的历史会话应隔离或清理，不能默认分配给任意用户。

## 测试

```bash
uv run pytest
```

## M-Chat 客户端联调

客户端的 Vite 开发代理已将 `/api` 转发到 `http://127.0.0.1:8090`。启动本服务后，在
`m-chat-client` 配置 `VITE_MCHAT_API_BASE_URL=/`，即可让前端从 Mock 模式切到真实 Chat
接口。

当前已实现普通聊天闭环和第一版模型可靠性能力：

- `GET/POST/PATCH/DELETE /api/v1/chat/sessions`
- `POST /api/v1/chat/messages`
- `POST /api/v1/chat/stream`，事件顺序为 `message_start -> token* -> message_end`
- 客户端可传 `qualityLevel=auto|fast|balanced|strong`，但具体模型仍由服务端白名单决定
- 服务端支持首 Token 前重试、备用模型降级、按模型限流和按模型熔断
- `degraded` 事件会告知客户端发生了模型降级，`message_end` 会返回实际模型和降级原因
- 首 Token 后发生中断时不会切换模型，也不会把半截回答保存为正常上下文
- `PATCH /api/v1/chat/sessions/{sessionId}` 返回会话摘要，不返回 `messages`

会话和消息已通过 ChatStore 端口接入 Redis，并按租户和用户隔离；服务重启后仍可恢复，多实例可以共享会话数据。模型
限流和熔断目前仍是单进程内存实现，多实例生产部署时应替换为 Redis 或网关能力。RAG
知识库接口暂仅返回空列表，客户端会自动按普通聊天模式请求模型。

## 提示词模板与变量填充

模型调用前会经过独立的 `PromptAssembler`。它负责将后端受控变量和会话历史渲染为
LangChain `BaseMessage` 列表；模型路由、重试、限流、熔断、降级等能力完全不感知模板内容。

```text
ChatStore 历史消息
  → PromptAssembler（模板选择、变量校验、历史截断）
  → ModelRequest.messages
  → ResilientModelInvoker（路由、重试、限流、熔断、降级）
  → Chat Model
```

当前内置 `normal_chat`、`complex_reasoning`、`tool_calling` 和 `rag_answer` 模板。普通模板会
填充应用名称、回答语言和任务说明；RAG 模板会把知识片段明确标识为“参考资料而非系统指令”，
并要求资料不足时不能编造。

企业级边界约束如下：

- 模板仅由后端 `app/prompts/templates.py` 注册，客户端不能提交任意模板文本或 `systemPrompt`。
- 动态变量由服务端 `PromptContext` 提供；当前来自环境配置。接入认证后，用户角色、租户、数据权限应由鉴权结果填入，而不是信任前端参数。
- 历史中的 `SystemMessage` 会被过滤，防止外部渠道导入的消息覆盖平台 System Prompt。
- `PROMPT_MAX_HISTORY_MESSAGES` 限制上下文规模，避免无限增长导致成本和首 Token 延迟上升。
- RAG 检索结果应由后续 Retriever 写入 `PromptContext.knowledge_context`；不要把前端上传的任意文本直接填入该字段。

新增业务场景时，按以下方式扩展：在 `app/prompts/templates.py` 注册受控模板，在
`PromptContext` 增加经过后端校验的字段，并由应用服务在调用 `PromptAssembler` 前填充。
不要在路由层拼接字符串，也不要把模板变量透传给前端。

## 设计说明

这个案例把代码分成三层：

- `main.py`：应用入口，创建 FastAPI 实例并挂载路由。
- `app/api/http_router.py`：HTTP 接口层，只处理 HTTP 请求、响应和错误转换。
- `app/schemas.py`：数据契约层，对应 Java 项目里的 DTO。
- `app/chat_generation.py`：聊天生成应用服务，组装提示词并编排模型调用。
- `app/llm/candidate_selection.py`：模型候选选择器，按质量等级生成主模型与降级模型的调用顺序；它不负责 HTTP 路由。

这样做的好处是后续扩展 AI Agent 能更自然：

- 加 RAG 时，可以在 service 层前面加入检索链路。
- 加工具调用时，可以把 tool binding 放到模型服务层。
- 加会话记忆时，可以在接口层和服务层之间增加 session/message store。
- 换模型供应商时，接口契约不用变，只替换 service 实现。

## 官方文档

- LangChain DeepSeek 集成：https://python.langchain.com/docs/integrations/chat/deepseek/
- DeepSeek API 文档：https://api-docs.deepseek.com/
- FastAPI 文档：https://fastapi.tiangolo.com/