fix(backend): wrap SSE astream_events in try/except/finally
Tool call failures (httpx network errors) propagated as unhandled exceptions
through LangGraph TaskGroup, killing SSE stream mid-flight. Frontend received
neither tokens nor done event, leaving it stuck.
- except: sends {"type":"error"} SSE event so frontend shows error message
- finally: persists partial AI content and always sends {"type":"done"}
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
gongzhiyong
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
e766564f34
commit
1e735202fc
@@ -4,94 +4,102 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
|
||||
|
||||
## Overview
|
||||
|
||||
so-c-chat-clone — 企业级 Gemini 风格对话系统，前后端分离。
|
||||
so-c-chat-clone — 企业级 Gemini 风格对话系统，前后端分离。基于 LangGraph ReAct Agent 编排，支持知识库检索、工单查询、外部搜索、文档生成、沙盒执行。
|
||||
|
||||
- **前端** (`frontend/`): Next.js 16 + React 19 + Tailwind CSS 4 + shadcn/ui，1:1 复刻 Google Gemini UI
|
||||
- **后端** (`backend/`): LangChain + LangGraph + Litestar，基于 LangGraph 编排的对话 Agent
|
||||
## Commands
|
||||
|
||||
## Project Structure
|
||||
|
||||
```
|
||||
├── frontend/ # Next.js 前端（前端代码未经明确指定不允许修改）
|
||||
├── backend/ # Python 后端（LangChain + LangGraph）
|
||||
├── gpthd.md # 后端功能方案（18项功能，开发前必读）
|
||||
├── EXTERNAL_SERVICES.md # 外部服务凭据与接入配置
|
||||
└── claudehd.md # 后端技术方案（按功能拆解）
|
||||
```
|
||||
|
||||
## Frontend
|
||||
|
||||
### Commands
|
||||
```bash
|
||||
cd frontend
|
||||
npm install && npm run dev # Dev server (localhost:3000)
|
||||
npm run build # Production build
|
||||
npm run lint # ESLint
|
||||
```
|
||||
|
||||
### Architecture
|
||||
- Entry: `frontend/app/page.tsx` → `<GeminiChat />`
|
||||
- `GeminiChat.tsx` owns all state, composes sidebar/topbar/input/message/welcome components
|
||||
- `GeminiInput.tsx` has `activeTools` (Set\<string\>) and `selectedModel` ("flash"|"pro") as local state
|
||||
- `simulateAIResponse()` is the mock function to be replaced by real backend API
|
||||
- All mock data (conversations, tickets) lives in `GeminiChat.tsx`
|
||||
- Dark theme only, hardcoded palette (`#131314` bg, `#1e1e1e` sidebar, `#4285f4→#a855f7` gradient)
|
||||
|
||||
## Backend
|
||||
|
||||
### Tech Stack
|
||||
- **Web**: Litestar + Uvicorn
|
||||
- **Graph**: LangGraph StateGraph + create_react_agent (ReAct)
|
||||
- **LLM**: LangChain AzureChatOpenAI (gpt-5.4)
|
||||
- **Tools**: LangChain @tool (KB search, Jina web search, Daytona sandbox, Doc Creator, Gongdan tickets)
|
||||
- **DB**: PostgreSQL + asyncpg + LangGraph AsyncPostgresSaver
|
||||
- **Cache**: Redis (Azure)
|
||||
- **Storage**: Azure Blob Storage
|
||||
- **Async**: Azure Service Bus
|
||||
|
||||
### Commands
|
||||
### Backend
|
||||
```bash
|
||||
cd backend
|
||||
python -m venv .venv && source .venv/bin/activate
|
||||
pip install -r requirements.txt # Or: pip install -e .
|
||||
pip install -r requirements.txt
|
||||
uvicorn app.main:app --port 8000 --reload # Dev server
|
||||
```
|
||||
|
||||
### Core API
|
||||
```
|
||||
POST /api/chat/stream # SSE streaming chat (replaces simulateAIResponse)
|
||||
GET /api/conversations # List conversations
|
||||
POST /api/conversations # Create conversation
|
||||
GET /api/tickets # Proxy to Gongdan ticket system
|
||||
GET /health # Health check
|
||||
### Frontend (read-only unless explicitly authorized)
|
||||
```bash
|
||||
cd frontend
|
||||
npm install && npm run dev # localhost:3000
|
||||
npm run build
|
||||
```
|
||||
|
||||
### Development Phases (see gpthd.md for full details)
|
||||
1. Basic chat graph + PostgreSQL + SSE streaming
|
||||
2. Tool integration (KB Agent, tickets, ReAct routing)
|
||||
3. External search (Jina Search/Reader/Rerank) + Redis cache
|
||||
4. Doc generation + Sandbox + Blob Storage + Service Bus
|
||||
### Deployment
|
||||
```bash
|
||||
# Manual deploy (Oryx builds dependencies on Azure)
|
||||
az webapp up --name soc-backend --resource-group Operation --runtime "PYTHON:3.12"
|
||||
|
||||
## External Services
|
||||
# Or push to main branch — GitHub Actions auto-deploys via .github/workflows/deploy-backend.yml
|
||||
git push origin main
|
||||
```
|
||||
|
||||
All credentials in `EXTERNAL_SERVICES.md`. Key services:
|
||||
## Architecture
|
||||
|
||||
| Service | Purpose |
|
||||
|---------|---------|
|
||||
| Azure OpenAI (gpt-5.4) | LLM generation |
|
||||
| KB Agent (Azure AI Search) | Internal knowledge retrieval |
|
||||
| Jina AI (Search/Reader/Rerank) | External web search |
|
||||
| Daytona | Sandboxed code execution |
|
||||
| Doc Creator Agent | Word/PPT/Excel generation |
|
||||
| Gongdan API | Ticket system (read-only) |
|
||||
| PostgreSQL (Azure) | Persistence |
|
||||
| Redis (Azure) | Caching |
|
||||
| Azure Blob Storage | File storage |
|
||||
| Azure Service Bus | Async task queue |
|
||||
### Request Flow
|
||||
```
|
||||
Frontend POST /api/chat/stream
|
||||
→ api/chat.py: resolve_tools() → get_chat_graph() → astream_events()
|
||||
→ graph/builder.py: create_react_agent (with tools) OR plain StateGraph (no tools)
|
||||
→ tools/*.py: LangChain @tool functions call external services
|
||||
→ SSE events: token / tool_start / tool_end / done
|
||||
```
|
||||
|
||||
### Graph Strategy (graph/builder.py)
|
||||
- **With tools**: `create_react_agent(llm, tools, checkpointer)` — ReAct pattern, LLM decides tool calls
|
||||
- **Without tools**: Simple `StateGraph(ChatState)` with single `call_model` node
|
||||
- Graphs cached by `(model, frozenset(tool_names))` to avoid re-compilation
|
||||
- Model presets: `flash` (max_tokens=500, temp=0.2), `pro` (max_tokens=4096, temp=0.3)
|
||||
|
||||
### Tool System (tools/__init__.py)
|
||||
Frontend sends tool keys → `resolve_tools()` maps to LangChain @tool objects → bound to ReAct agent.
|
||||
|
||||
| Frontend Key | Tools | External Service |
|
||||
|-------------|-------|-----------------|
|
||||
| `"knowledge"` | `kb_search` | KB Agent (Azure AI Search) |
|
||||
| `"tickets"` | `ticket_list`, `ticket_detail` | Gongdan API |
|
||||
| `"search"` | `web_search` | Jina Search/Reader/Rerank |
|
||||
| `"document"` | `generate_document` | Doc Creator Agent |
|
||||
| `"sandbox"` | `sandbox_run` | Daytona API |
|
||||
|
||||
### Data Layer
|
||||
- **PostgreSQL** (`store/postgres.py`): SQLAlchemy async ORM — `Conversation` and `Message` models, auto-creates tables on startup
|
||||
- **LangGraph Checkpointer** (`store/memory.py`): `AsyncPostgresSaver` via psycopg (separate connection from SQLAlchemy, uses `sslmode=require` not `ssl=require`)
|
||||
- **Redis** (`cache/redis.py`): Search result caching, key=`search:{hash}:{model}`, TTL=300s, graceful degradation on failure
|
||||
- **Azure Blob Storage** (`storage/blob.py`): File uploads/downloads for attachments and generated docs
|
||||
- **Azure Service Bus** (`tasks/bus.py`): Async task dispatch for long-running operations
|
||||
|
||||
### SSE Event Protocol
|
||||
```json
|
||||
{"type": "token", "content": "..."} // Streamed text chunk
|
||||
{"type": "tool_start", "tool": "kb_search"}
|
||||
{"type": "tool_end", "tool": "kb_search"}
|
||||
{"type": "done"} // End of stream
|
||||
```
|
||||
|
||||
### Lifespan (main.py)
|
||||
Startup creates DB tables. Shutdown closes: checkpointer → Redis → Blob client → Service Bus → SQLAlchemy engine.
|
||||
|
||||
## Configuration
|
||||
|
||||
All config via `pydantic-settings` in `app/config.py`, reads from `.env` file. Full credentials reference in `EXTERNAL_SERVICES.md`.
|
||||
|
||||
Key env vars: `AZURE_OPENAI_*`, `DATABASE_URL`, `KB_AGENT_*`, `GONGDAN_*`, `JINA_API_KEY`, `REDIS_URL`, `DOC_AGENT_*`, `DAYTONA_*`, `AZURE_STORAGE_CONNECTION_STRING`, `AZURE_SERVICE_BUS_CONNECTION_STRING`.
|
||||
|
||||
## Deployment
|
||||
|
||||
- **Azure Web App**: `soc-backend` in `Operation` resource group, Python 3.12, B1 Linux
|
||||
- **Startup**: `gunicorn -w 2 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 --timeout 120 app.main:app`
|
||||
- **Critical setting**: `WEBSITES_PORT=8000` (Azure defaults to 8080)
|
||||
- **CI/CD**: GitHub Actions with OIDC auth (`azure/login@v2`) → zip deploy, triggers on `backend/**` changes to main
|
||||
- **Always On**: Enabled to avoid cold starts
|
||||
|
||||
## Constraints
|
||||
|
||||
- **Frontend code is read-only** unless explicitly authorized. Only approved change: extending `onSubmit` to pass `tools[]` and `model` to backend.
|
||||
- **Azure resources** must stay within `AuthData` and `Operation` resource groups only.
|
||||
- **CI/CD** is managed by the user, not by agents.
|
||||
- **Tool invocation policy**: User-selected tools are passed to the LangGraph ReAct Agent as available tools. The Agent decides whether to actually use them. If it decides not to, it must explain why in its response.
|
||||
- **Frontend is read-only** unless explicitly authorized
|
||||
- **Azure resources**: `Operation` and `AuthData` resource groups only — all `az` commands must include `--resource-group`
|
||||
- **Tool invocation policy**: User-selected tools are passed to ReAct Agent as available. Agent decides whether to call them. If it decides not to, it must explain why.
|
||||
- **Search depth**: `flash` = quick (top 3, no Reader/Rerank), `pro` = deep (top 10, concurrent Reader, Rerank top 5)
|
||||
|
||||
## Known Issues
|
||||
|
||||
- `GET /api/tickets/{id}` returns 404 — upstream Gongdan API issue, not backend bug
|
||||
- `config.py` has hardcoded DB/Redis passwords as defaults — should be empty strings (security risk if repo goes public)
|
||||
|
||||
+51
-39
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
@@ -76,51 +77,62 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
|
||||
}
|
||||
|
||||
full_content: list[str] = []
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
async for event in graph.astream_events(
|
||||
input_data,
|
||||
config=config,
|
||||
version="v2",
|
||||
):
|
||||
kind = event.get("event", "")
|
||||
try:
|
||||
async for event in graph.astream_events(
|
||||
input_data,
|
||||
config=config,
|
||||
version="v2",
|
||||
):
|
||||
kind = event.get("event", "")
|
||||
|
||||
if kind == "on_chat_model_stream":
|
||||
chunk = event.get("data", {}).get("chunk")
|
||||
if chunk and hasattr(chunk, "content") and chunk.content:
|
||||
# Only stream text content, skip tool call chunks
|
||||
if isinstance(chunk.content, str):
|
||||
full_content.append(chunk.content)
|
||||
sse_data = json.dumps(
|
||||
{"type": "token", "content": chunk.content},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
if kind == "on_chat_model_stream":
|
||||
chunk = event.get("data", {}).get("chunk")
|
||||
if chunk and hasattr(chunk, "content") and chunk.content:
|
||||
# Only stream text content, skip tool call chunks
|
||||
if isinstance(chunk.content, str):
|
||||
full_content.append(chunk.content)
|
||||
sse_data = json.dumps(
|
||||
{"type": "token", "content": chunk.content},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
|
||||
elif kind == "on_tool_start":
|
||||
# Notify frontend that a tool is being called
|
||||
tool_name = event.get("name", "unknown")
|
||||
sse_data = json.dumps(
|
||||
{"type": "tool_start", "tool": tool_name},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
elif kind == "on_tool_start":
|
||||
# Notify frontend that a tool is being called
|
||||
tool_name = event.get("name", "unknown")
|
||||
sse_data = json.dumps(
|
||||
{"type": "tool_start", "tool": tool_name},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
|
||||
elif kind == "on_tool_end":
|
||||
tool_name = event.get("name", "unknown")
|
||||
sse_data = json.dumps(
|
||||
{"type": "tool_end", "tool": tool_name},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
elif kind == "on_tool_end":
|
||||
tool_name = event.get("name", "unknown")
|
||||
sse_data = json.dumps(
|
||||
{"type": "tool_end", "tool": tool_name},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {sse_data}\n\n".encode("utf-8")
|
||||
|
||||
# Persist AI response
|
||||
ai_content = "".join(full_content)
|
||||
if ai_content:
|
||||
await _persist_ai_message(request.conversation_id, ai_content)
|
||||
except Exception as exc:
|
||||
logger.error("SSE stream error for conversation %s: %s", request.conversation_id, exc, exc_info=True)
|
||||
error_data = json.dumps(
|
||||
{"type": "error", "content": "请求处理出现错误，请重试"},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"data: {error_data}\n\n".encode("utf-8")
|
||||
|
||||
# Send done signal
|
||||
done_data = json.dumps({"type": "done"})
|
||||
yield f"data: {done_data}\n\n".encode("utf-8")
|
||||
finally:
|
||||
# Persist whatever AI content was streamed before the error (if any)
|
||||
ai_content = "".join(full_content)
|
||||
if ai_content:
|
||||
await _persist_ai_message(request.conversation_id, ai_content)
|
||||
|
||||
# Always send done so the frontend closes the stream cleanly
|
||||
done_data = json.dumps({"type": "done"})
|
||||
yield f"data: {done_data}\n\n".encode("utf-8")
|
||||
|
||||
|
||||
@post("/api/chat/stream")
|
||||
|
||||
Reference in New Issue
Block a user