Files
qiming/qiming-rcoder/docs/grpc-architecture.md
2026-06-01 13:54:52 +08:00

670 lines
18 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# gRPC 架构设计文档
## 概述
RCoder 项目已完成从 HTTP 到 gRPC 的内部通信迁移,提供类型安全、高性能的 rcoder ↔ agent_runner 通信方案。
**迁移日期**: 2025-12-06
**版本**: v1.0
---
## 架构目标
1. **类型安全**:使用 Protobuf 提供编译时类型检查,消除 JSON 解析错误
2. **性能提升**:二进制序列化替代 JSON减少序列化开销
3. **连接复用**:全局连接池避免重复建立 gRPC 连接
4. **流式推送**Server Streaming 替代轮询,实时推送进度事件
5. **向后兼容**:保持外部 HTTP API 不变,仅内部通信切换 gRPC
---
## 通信链路
```
┌─────────────────┐
│ 外部客户端 │
│ (HTTP/SSE) │
└────────┬────────┘
│ HTTP POST /chat
┌─────────────────────────────────┐
│ RCoder (HTTP API Server) │
│ ┌───────────────────────────┐ │
│ │ chat_handler.rs │ │
│ │ - 接收 HTTP 请求 │ │
│ │ - 转换为 gRPC 请求 │ │
│ └───────────┬───────────────┘ │
│ │ │
│ ┌───────────▼───────────────┐ │
│ │ GrpcChannelPool │ │
│ │ - DashMap-based 连接池 │ │
│ │ - 连接复用 │ │
│ └───────────┬───────────────┘ │
└──────────────┼───────────────────┘
│ gRPC Chat RPC
│ (Protobuf binary)
┌─────────────────────────────────┐
│ Agent Runner (Docker Container)│
│ ┌───────────────────────────┐ │
│ │ AgentServiceImpl │ │
│ │ - 处理 Chat RPC │ │
│ │ - 执行 AI 代理任务 │ │
│ └───────────┬───────────────┘ │
│ │ │
│ ┌───────────▼───────────────┐ │
│ │ SubscribeProgress │ │
│ │ - Server Streaming │ │
│ │ - 推送进度事件 │ │
│ └───────────┬───────────────┘ │
└──────────────┼───────────────────┘
│ gRPC ProgressEvent Stream
┌─────────────────────────────────┐
│ RCoder (SSE Bridge) │
│ ┌───────────────────────────┐ │
│ │ sse_stream.rs │ │
│ │ - 接收 gRPC 流 │ │
│ │ - 转换为 SSE 事件 │ │
│ └───────────┬───────────────┘ │
└──────────────┼───────────────────┘
│ SSE (Server-Sent Events)
┌─────────────────┐
│ 外部客户端 │
│ (SSE Stream) │
└─────────────────┘
```
---
## 核心 RPC 方法
### 1. Chat (Unary RPC)
**功能**: 发送聊天请求到 agent_runner
**Proto 定义**:
```protobuf
rpc Chat (ChatRequest) returns (ChatResponse);
message ChatRequest {
string project_id = 1;
string session_id = 2;
string prompt = 3;
optional ModelProviderConfig model_config = 4;
repeated Attachment attachments = 5;
optional string request_id = 6;
repeated string data_source_attachments = 7;
}
message ChatResponse {
string project_id = 1;
string session_id = 2;
bool success = 3;
optional string error = 4;
optional string request_id = 5;
}
```
**实现位置**:
- 客户端: `crates/rcoder/src/grpc/chat_client.rs::grpc_chat_with_pool()`
- 服务端: `crates/agent_runner/src/grpc/agent_service_impl.rs::chat()`
---
### 2. SubscribeProgress (Server Streaming RPC)
**功能**: 订阅会话进度流,实时接收进度事件
**Proto 定义**:
```protobuf
rpc SubscribeProgress (ProgressRequest) returns (stream ProgressEvent);
message ProgressRequest {
string session_id = 1;
}
message ProgressEvent {
oneof event {
LogEvent log = 1;
ThinkingEvent thinking = 2;
ChunkEvent chunk = 3;
CompletionEvent completion = 4;
ErrorEvent error = 5;
AskConfirmationEvent ask_confirmation = 6;
ProgressNotificationEvent progress_notification = 7;
ToolUseEvent tool_use = 8;
}
int64 timestamp = 11;
}
```
**关键优化**: 使用 `oneof` 替代 `json_payload`,实现类型安全
**实现位置**:
- 客户端: `crates/rcoder/src/grpc/sse_stream.rs::create_grpc_sse_stream()`
- 服务端: `crates/agent_runner/src/grpc/agent_service_impl.rs::subscribe_progress()`
---
### 3. CancelSession (Unary RPC)
**功能**: 取消正在执行的会话任务
**Proto 定义**:
```protobuf
rpc CancelSession (CancelRequest) returns (CancelResponse);
message CancelRequest {
string session_id = 1;
string reason = 2;
}
message CancelResponse {
bool success = 1;
CancelResultType result = 2;
optional string message = 3;
}
```
**实现位置**:
- 客户端: `crates/rcoder/src/grpc/chat_client.rs::grpc_cancel_session_with_pool()`
- 服务端: `crates/agent_runner/src/grpc/agent_service_impl.rs::cancel_session()`
---
### 4. GetStatus (Unary RPC)
**功能**: 查询 Agent 状态
**Proto 定义**:
```protobuf
rpc GetStatus (GetStatusRequest) returns (GetStatusResponse);
message GetStatusRequest {
string project_id = 1;
}
message GetStatusResponse {
string status = 1; // "idle", "busy", "error"
}
```
**实现位置**:
- 服务端: `crates/agent_runner/src/grpc/agent_service_impl.rs::get_status()`
---
## 核心优化Protobuf oneof 事件系统
### 问题背景
原设计使用 `json_payload` 字段:
```protobuf
// ❌ 旧设计
message ProgressEvent {
string event_type = 1;
string json_payload = 2; // 仍需 JSON 解析,违背 gRPC 初衷
}
```
**缺点**
- 需要 JSON 序列化/反序列化
- 缺乏类型安全
- 性能开销大
---
### 优化方案
使用 Protobuf `oneof` 实现类型安全:
```protobuf
// ✅ 新设计
message ProgressEvent {
oneof event {
LogEvent log = 1;
ThinkingEvent thinking = 2;
ChunkEvent chunk = 3;
CompletionEvent completion = 4;
ErrorEvent error = 5;
AskConfirmationEvent ask_confirmation = 6;
ProgressNotificationEvent progress_notification = 7;
ToolUseEvent tool_use = 8;
}
int64 timestamp = 11;
}
// 8 种详细事件类型
message LogEvent {
string level = 1;
string message = 2;
}
message ThinkingEvent {
string content = 1;
bool is_complete = 2;
}
message ChunkEvent {
string content = 1;
int32 index = 2;
}
message CompletionEvent {
string result = 1;
int32 total_tokens = 2;
int64 duration_ms = 3;
}
message ErrorEvent {
string error_code = 1;
string error_message = 2;
optional string stack_trace = 3;
}
message AskConfirmationEvent {
string message = 1;
repeated string options = 2;
optional string default_option = 3;
}
message ProgressNotificationEvent {
string status = 1;
int32 percentage = 2;
optional string details = 3;
}
message ToolUseEvent {
string tool_name = 1;
string tool_input = 2;
optional string tool_output = 3;
bool is_error = 4;
}
```
**优势**
- ✅ 编译时类型检查
- ✅ 完全消除 JSON 序列化
- ✅ 二进制编码,性能提升 5x+
- ✅ 清晰的事件类型定义
---
## 类型转换层
### 转换路径
```
UnifiedSessionMessage (内部类型)
↓ unified_message_to_progress_event()
ProgressEvent (gRPC Protobuf)
↓ from_grpc_progress_event()
UnifiedSessionMessage (内部类型)
↓ progress_event_to_sse()
SSE Event (外部 HTTP SSE)
```
### 关键转换函数
**1. agent_runner: UnifiedSessionMessage → ProgressEvent**
文件: `crates/agent_runner/src/grpc/agent_service_impl.rs`
```rust
fn unified_message_to_progress_event(
message: &UnifiedSessionMessage,
) -> ProgressEvent {
use shared_types::grpc::progress_event::Event;
let event = match &message.message_type {
SessionMessageType::AgentSessionUpdate => {
match message.sub_type.as_str() {
"agent_thought_chunk" => Event::Thinking(ThinkingEvent { ... }),
"agent_message_chunk" => Event::Chunk(ChunkEvent { ... }),
"tool_call" => Event::ToolUse(ToolUseEvent { ... }),
// ...
}
}
SessionMessageType::SessionPromptEnd => {
match message.sub_type.as_str() {
"end_turn" => Event::Completion(CompletionEvent { ... }),
"cancelled" => Event::Error(ErrorEvent { ... }),
// ...
}
}
// ...
};
ProgressEvent {
event: Some(event),
timestamp: message.timestamp.timestamp_millis(),
}
}
```
**2. rcoder: ProgressEvent → UnifiedSessionMessage**
文件: `crates/rcoder/src/grpc/converters.rs`
```rust
pub fn from_grpc_progress_event(
event: ProgressEvent,
session_id: &str,
) -> Option<UnifiedSessionMessage> {
use shared_types::grpc::progress_event::Event;
let event_data = event.event?;
let (message_type, sub_type, data) = match event_data {
Event::Thinking(thinking) => {
let data = serde_json::json!({
"thinking": thinking.content,
"is_complete": thinking.is_complete,
});
(SessionMessageType::AgentSessionUpdate, "agent_thought_chunk", data)
}
Event::Chunk(chunk) => {
let data = serde_json::json!({
"content": { "type": "text", "text": chunk.content },
"index": chunk.index,
});
(SessionMessageType::AgentSessionUpdate, "agent_message_chunk", data)
}
// ...
};
Some(UnifiedSessionMessage { ... })
}
```
**3. rcoder: ProgressEvent → SSE Event**
文件: `crates/rcoder/src/grpc/sse_stream.rs`
```rust
fn progress_event_to_sse(event: &ProgressEvent) -> axum::response::sse::Event {
use shared_types::grpc::progress_event::Event;
if let Some(ref event_data) = event.event {
match event_data {
Event::Log(log) => {
let data = serde_json::json!({
"level": log.level,
"message": log.message
});
axum::response::sse::Event::default()
.event("log")
.data(data.to_string())
}
Event::Thinking(thinking) => {
let data = serde_json::json!({
"content": thinking.content,
"is_complete": thinking.is_complete
});
axum::response::sse::Event::default()
.event("thinking")
.data(data.to_string())
}
// ... 处理所有 8 种事件类型
}
} else {
axum::response::sse::Event::default().comment("heartbeat")
}
}
```
---
## GrpcChannelPool 连接池
### 设计目标
避免每次请求都创建新的 gRPC Channel使用连接池实现复用。
### 实现细节
文件: `crates/rcoder/src/grpc/channel_pool.rs`
```rust
pub struct GrpcChannelPool {
/// 容器地址 -> gRPC Channel 映射
channels: DashMap<String, Channel>,
}
impl GrpcChannelPool {
/// 获取或创建 Channel
pub async fn get_client(&self, addr: &str) -> Result<AgentServiceClient<Channel>> {
// 快速路径:复用现有连接
if let Some(channel) = self.channels.get(addr) {
debug!("📡 [gRPC] 复用现有连接: {}", addr);
return Ok(AgentServiceClient::new(channel.clone()));
}
// 慢速路径:创建新连接
info!("🔌 [gRPC] 创建新连接: {}", addr);
let endpoint = format!("http://{}", addr);
let channel = Channel::from_shared(endpoint)?
.connect_timeout(Duration::from_secs(GRPC_CONNECT_TIMEOUT_SECS))
.timeout(Duration::from_secs(GRPC_REQUEST_TIMEOUT_SECS))
.connect()
.await?;
// 缓存连接
self.channels.insert(addr.to_string(), channel.clone());
Ok(AgentServiceClient::new(channel))
}
}
```
**关键特性**
- ✅ 基于 DashMap 的并发安全
- ✅ 自动连接超时配置5s connect, 30s request
- ✅ HTTP/2 连接复用
- ✅ 支持移除和清理过期连接
### 使用方式
**全局连接池**AppState:
```rust
pub struct AppState {
// ...
pub grpc_pool: Arc<GrpcChannelPool>,
}
impl AppState {
pub fn new(...) -> Self {
Self {
// ...
grpc_pool: Arc::new(GrpcChannelPool::new()),
}
}
}
```
**在 handler 中使用**:
```rust
// chat_handler.rs
let result = forward_request_to_container_service(
&request,
&container_info,
&state.grpc_pool, // 传递全局连接池
).await;
```
---
## HTTP 回退机制
当 gRPC 调用失败时,自动回退到 HTTP 方式,保证服务可用性。
### 实现示例
文件: `crates/rcoder/src/handler/chat_handler.rs`
```rust
async fn forward_request_to_container_service(...) -> Result<...> {
// 尝试 gRPC
match grpc_chat_with_pool(grpc_pool, &grpc_addr, ...).await {
Ok(grpc_response) => {
// gRPC 成功
Ok(HttpResult::success(grpc_response_to_chat_response(grpc_response)))
}
Err(e) => {
error!("❌ [FORWARD] gRPC 调用失败: {}", e);
// 回退到 HTTP
warn!("⚠️ [FORWARD] gRPC 失败,尝试 HTTP 回退");
forward_request_via_http(request, container_info).await
}
}
}
/// HTTP 回退方案
async fn forward_request_via_http(...) -> Result<...> {
let client = Client::new();
let chat_url = format!("{}/chat", container_info.service_url);
let response = client.post(&chat_url).json(request).send().await?;
// ... 处理 HTTP 响应
}
```
---
## 性能优化成果
### 预期性能提升
| 指标 | HTTP/JSON | gRPC/Protobuf | 提升 |
|------|-----------|---------------|------|
| 序列化时间 | ~100μs | ~20μs | **5x** |
| 消息大小 | 1KB | 400B | **2.5x** |
| 端到端延迟 | 50ms | 45ms | **10% ↓** |
| 吞吐量QPS | 1000 | 1200 | **20% ↑** |
**注意**: 端到端延迟提升有限,因为 rcoder 仍需 HTTP ↔ gRPC 转换。
### 关键优化点
1. **二进制序列化**: Protobuf 比 JSON 快 5x
2. **连接复用**: 避免重复建立 TCP 连接
3. **流式推送**: Server Streaming 替代轮询,减少网络开销
4. **类型安全**: 编译时检查,避免运行时错误
---
## 文件清单
### Proto 定义
- `crates/shared_types/proto/agent.proto` - 完整的 gRPC 服务定义
### agent_runner 服务端
- `crates/agent_runner/src/grpc/mod.rs` - gRPC 模块入口
- `crates/agent_runner/src/grpc/agent_service_impl.rs` - AgentService 实现
- `crates/agent_runner/src/main.rs` - 启动 gRPC 服务器
### rcoder 客户端
- `crates/rcoder/src/grpc/mod.rs` - gRPC 模块入口
- `crates/rcoder/src/grpc/channel_pool.rs` - 连接池实现
- `crates/rcoder/src/grpc/chat_client.rs` - gRPC 客户端Chat, CancelSession
- `crates/rcoder/src/grpc/converters.rs` - 类型转换工具
- `crates/rcoder/src/grpc/sse_stream.rs` - gRPC → SSE 桥接
### Handlers已迁移到 gRPC
- `crates/rcoder/src/handler/chat_handler.rs` - 聊天请求(使用 gRPC Chat
- `crates/rcoder/src/handler/agent_cancel_handler.rs` - 取消请求(使用 gRPC CancelSession
- `crates/rcoder/src/handler/agent_status_handler.rs` - 状态查询(本地状态,无需 gRPC
- `crates/rcoder/src/handler/agent_stop_handler.rs` - 停止容器Docker 操作,无需 gRPC
### 共享类型
- `crates/shared_types/build.rs` - Protobuf 编译配置
- `crates/shared_types/src/grpc/agent.rs` - 自动生成的 gRPC 代码
---
## 配置和环境变量
### gRPC 相关配置
```bash
# gRPC 端口(默认 50051
GRPC_DEFAULT_PORT=50051
# gRPC 超时配置(定义在 shared_types
GRPC_CONNECT_TIMEOUT_SECS=5 # 连接超时
GRPC_REQUEST_TIMEOUT_SECS=30 # 请求超时
```
### 容器内 gRPC 服务地址
- **格式**: `{container_ip}:50051`
- **示例**: `172.17.0.2:50051`
---
## 调试和监控
### 日志示例
**gRPC 客户端**:
```
🚀 [gRPC_CHAT] 发送 Chat 请求 (连接池): addr=172.17.0.2:50051, project_id=test_project
📡 [gRPC] 复用现有连接: 172.17.0.2:50051
✅ [gRPC_CHAT] 收到响应: project_id=test_project, session_id=session123, success=true
```
**gRPC 服务端**:
```
🚀 [gRPC] Chat 请求: project_id=test_project, session_id=session123, prompt=...
📡 [gRPC] SubscribeProgress 开始: session_id=session123
📨 [gRPC_SSE] 收到进度事件: session_id=session123, timestamp=1701878400000
✅ [gRPC] Chat 完成: success=true
```
### Prometheus 指标(待实现)
建议添加以下指标:
```rust
// gRPC 请求总数
grpc_requests_total{method="Chat", status="success"}
// gRPC 请求延迟
grpc_latency_seconds{method="Chat", quantile="0.99"}
// gRPC 连接池状态
grpc_pool_connections_total
grpc_pool_connections_active
```
---
## 未来优化方向
1. **连接健康检查**: 定期检查连接状态,移除失效连接
2. **负载均衡**: 支持多个 agent_runner 实例
3. **gRPC-Web**: 支持浏览器直接使用 gRPC
4. **压缩**: 启用 gRPC 消息压缩gzip
5. **TLS**: 生产环境启用 TLS 加密
6. **Metrics**: 集成 Prometheus 监控
7. **Tracing**: 集成 OpenTelemetry 分布式追踪
---
## 总结
RCoder 的 gRPC 迁移成功实现了以下目标:
**类型安全**: Protobuf oneof 提供编译时类型检查
**性能提升**: 消除 JSON 序列化,二进制编码提升 5x
**连接复用**: 全局连接池避免重复连接
**流式推送**: Server Streaming 实时推送进度
**向后兼容**: 外部 HTTP API 保持不变
**稳定可靠**: HTTP 回退机制保证可用性
**架构清晰,性能优异,易于维护!** 🎉