From 8e986a7bb456701136004a06f233dd7c6e9d5be8 Mon Sep 17 00:00:00 2001 From: TetrisGGBOBot Date: Tue, 31 Mar 2026 14:14:48 +0000 Subject: [PATCH] refactor: unify stream API and remove sync search endpoint Use /search/stream as the single entry for both fast and deep modes via request params, suppress COT events in fast mode while keeping deep COT, and return 410 for deprecated /search. Made-with: Cursor --- README.md | 53 +++-------- ai_search_agent/agent.py | 197 +++++++++++++++++++++------------------ ai_search_agent/api.py | 32 ++----- function_app.py | 4 +- 4 files changed, 127 insertions(+), 159 deletions(-) diff --git a/README.md b/README.md index ab562ad..7903e5c 100644 --- a/README.md +++ b/README.md @@ -102,11 +102,10 @@ uvicorn serve:app --host 0.0.0.0 --port 8080 --reload | 方法 | 路径 | 说明 | |------|------|------| | GET | `/health` | 健康检查（本地 FastAPI） | -| POST | `/search` | 同步搜索（阻塞式，本地 FastAPI） | -| POST | `/search/stream` | 流式搜索（SSE 事件，仅常驻 FastAPI 服务） | +| POST | `/search/stream` | 统一流式搜索入口（SSE） | -> Azure Functions 部署时由 `function_app.py` 挂载到 `/api` 前缀，仅提供 `/api/health`、`/api/search`。 -> `/search/stream` 必须走常驻 `uvicorn` 服务（避免 Functions 网关缓冲导致假流式）。 +> Azure Functions 部署时由 `function_app.py` 挂载到 `/api` 前缀，对外路径分别为 `/api/health`、`/api/search/stream`。 +> 快速与深度通过请求参数 `search_mode` 区分，不再通过不同 URL 区分。 ### 1. 健康检查 @@ -115,10 +114,10 @@ curl https://aisousuo.azurewebsites.net/api/health # {"status": "ok"} ``` -### 2. 同步搜索 +### 2. 统一流式搜索 ```bash -curl -X POST https://aisousuo.azurewebsites.net/api/search \ +curl -N -X POST https://aisousuo.azurewebsites.net/api/search/stream \ -H "Content-Type: application/json" \ -d '{ "query": "PydanticAI 适合哪些场景？", @@ -128,37 +127,20 @@ curl -X POST https://aisousuo.azurewebsites.net/api/search \ ``` `search_mode` 说明： -- `fast`：30 秒时间预算，超时返回 `504` -- `deep`：120 秒时间预算，超时返回 `504` -- 两种模式均不做降级回答 -- `fast` 使用极简路径：跳过 `decompose/expand`，固定单轮搜索，Reader 最多读取 3 页，并关闭 `reranker/KG/confidence retry` +- `fast`：30 秒时间预算，超时事件为 `timeout`（`error=TIMEOUT_FAST_SEARCH`） +- `deep`：120 秒时间预算，超时事件为 `timeout`（`error=TIMEOUT_DEEP_SEARCH`） +- `fast`：无 COT 过程事件，只输出必要流式事件（如 `connected`、`llm_token`、`result`、`timeout`、`audit`） +- `deep`：输出完整 COT 过程事件 +- `fast` 继续使用极简路径：跳过 `decompose/expand`，固定单轮搜索，Reader 最多读取 3 页，并关闭 `reranker/KG/confidence retry` Jina 搜索结果固定上限： - 每次搜索查询最多保留 10 条结果用于后续处理 -响应包含： -- `answer.confidence` - 置信度评分 -- `answer.follow_up_questions` - 3 条推荐追问 -- `answer.citations[].trust_score` - 引用可信度 -- `search_rounds` - 多轮搜索轨迹 -- `knowledge_triples` - 抽取的知识三元组 -- `cache_hit` - 是否缓存命中 -- `audit_id` - 审计追踪 ID - -### 3. SSE 流式搜索（常驻 FastAPI） - -```bash -curl -N -X POST https://your-stream-service/search/stream \ - -H "Content-Type: application/json" \ - -d '{"query":"PydanticAI 适合哪些场景?","search_mode":"deep"}' -``` - 事件类型（含前端 COT 过程事件）： `connected`, `mode_budget`, `progress`, `query_plan`, `cache_hit`, `decomposed`, `expanded`, `round_started`, `search_hits`, `page_fetch_status`, `reasoning_note`, `round_finished`, `llm_started`, `llm_token`, `final_meta`, `result`, `timeout`, `audit` > 当请求超时时，SSE 会发送 `timeout` 事件并结束流。 - -> 如果你误调 Azure Functions 的 `/api/search/stream`，会返回 `501`，并提示使用独立流式服务地址。 +> `POST /search` 已下线，调用会返回 `410`，请统一改为 `/search/stream`。 --- @@ -236,19 +218,6 @@ func azure functionapp publish aisousuo --python curl https://aisousuo.azurewebsites.net/api/health ``` -### 流式服务部署（推荐 App Service / Container） - -```bash -# 启动常驻服务（示例） -uvicorn serve:app --host 0.0.0.0 --port 8080 -``` - -- 将前端流式请求改为 `https:///search/stream` -- Azure Functions 继续承载同步接口：`/api/search` -- 可在 Functions 环境变量中配置 `STREAM_SERVICE_BASE_URL=https://`，用于错误提示回传目标流式地址 - ---- - ## 设计架构 ### 数据流 diff --git a/ai_search_agent/agent.py b/ai_search_agent/agent.py index 00ef0f4..ec6a3f9 100644 --- a/ai_search_agent/agent.py +++ b/ai_search_agent/agent.py @@ -59,29 +59,31 @@ class AISearchAgent: ) -> AsyncGenerator[dict, None]: mode = self.resolve_search_mode(request) is_fast = mode == "fast" + emit_cot = not is_fast timeout_seconds = self.get_time_budget_seconds(request) max_rounds = 1 if is_fast else self.settings.max_search_rounds top_k = request.top_k_pages or self.settings.top_k_pages max_chars = request.max_page_chars or self.settings.max_page_chars cache_query = f"{mode}:{request.query}" - yield { - "type": "mode_budget", - "data": { - "search_mode": mode, - "deadline_sec": timeout_seconds, - "max_rounds": max_rounds, - "max_hits": self.MAX_SEARCH_HITS, - }, - } - yield { - "type": "progress", - "data": { - "stage": "planning", - "message": "开始分析查询并规划搜索。", - "progress": 5, - }, - } + if emit_cot: + yield { + "type": "mode_budget", + "data": { + "search_mode": mode, + "deadline_sec": timeout_seconds, + "max_rounds": max_rounds, + "max_hits": self.MAX_SEARCH_HITS, + }, + } + yield { + "type": "progress", + "data": { + "stage": "planning", + "message": "开始分析查询并规划搜索。", + "progress": 5, + }, + } # ── 0. Cache hit ──────────────────────────────────────────────── if self.settings.enable_cache: @@ -89,7 +91,8 @@ class AISearchAgent: if cached: cached["cache_hit"] = True cached["search_mode"] = mode - yield {"type": "cache_hit", "data": {"query": request.query}} + if emit_cot: + yield {"type": "cache_hit", "data": {"query": request.query}} yield {"type": "result", "data": cached} return @@ -107,7 +110,8 @@ class AISearchAgent: sub_queries = [request.query] expanded = [request.query] seen_q.add(request.query) - yield {"type": "decomposed", "data": {"sub_queries": sub_queries}} + if emit_cot: + yield {"type": "decomposed", "data": {"sub_queries": sub_queries}} else: sub_queries = await self.planner.decompose(request.query) yield {"type": "decomposed", "data": {"sub_queries": sub_queries}} @@ -121,14 +125,15 @@ class AISearchAgent: expanded.append(q) if len(expanded) > len(sub_queries): yield {"type": "expanded", "data": {"queries": expanded}} - yield { - "type": "query_plan", - "data": { - "original_query": request.query, - "sub_queries": sub_queries, - "expanded_queries": expanded, - }, - } + if emit_cot: + yield { + "type": "query_plan", + "data": { + "original_query": request.query, + "sub_queries": sub_queries, + "expanded_queries": expanded, + }, + } # ── 4. Multi-round search loop ─────────────────────────────────── all_results: list[SearchResult] = [] @@ -141,19 +146,20 @@ class AISearchAgent: round_index = 0 for round_index in range(1, max_rounds + 1): - yield { - "type": "round_started", - "data": {"round_index": round_index, "queries": current_queries}, - } - yield { - "type": "progress", - "data": { - "stage": "searching", - "message": f"第 {round_index} 轮搜索中。", - "progress": min(20 + round_index * 15, 55), - "round_index": round_index, - }, - } + if emit_cot: + yield { + "type": "round_started", + "data": {"round_index": round_index, "queries": current_queries}, + } + yield { + "type": "progress", + "data": { + "stage": "searching", + "message": f"第 {round_index} 轮搜索中。", + "progress": min(20 + round_index * 15, 55), + "round_index": round_index, + }, + } round_results: list[SearchResult] = [] for q in current_queries: @@ -165,31 +171,34 @@ class AISearchAgent: seen_urls.add(r.url) round_results.append(r) except Exception as exc: - yield {"type": "search_error", "data": {"query": q, "error": str(exc)}} - yield { - "type": "search_hits", - "data": { - "round_index": round_index, - "queries": current_queries, - "hits": [item.model_dump() for item in round_results[: self.MAX_SEARCH_HITS]], - }, - } + if emit_cot: + yield {"type": "search_error", "data": {"query": q, "error": str(exc)}} + if emit_cot: + yield { + "type": "search_hits", + "data": { + "round_index": round_index, + "queries": current_queries, + "hits": [item.model_dump() for item in round_results[: self.MAX_SEARCH_HITS]], + }, + } page_limit = self.FAST_MAX_READ_PAGES if is_fast else self.MAX_SEARCH_HITS pages = await self.jina.read_pages( round_results[:page_limit], max_page_chars=max_chars ) usable = [p for p in pages if p.fetched and p.usable] - yield { - "type": "page_fetch_status", - "data": { - "round_index": round_index, - "total": len(pages), - "done": len(pages), - "success": len([p for p in pages if p.fetched]), - "failed": len([p for p in pages if not p.fetched]), - }, - } + if emit_cot: + yield { + "type": "page_fetch_status", + "data": { + "round_index": round_index, + "total": len(pages), + "done": len(pages), + "success": len([p for p in pages if p.fetched]), + "failed": len([p for p in pages if not p.fetched]), + }, + } if (not is_fast) and self.settings.enable_reranker and len(usable) > 1: try: @@ -215,14 +224,15 @@ class AISearchAgent: all_pages.extend(pages) all_usable.extend(usable) - yield {"type": "round_finished", "data": trace.model_dump()} - yield { - "type": "reasoning_note", - "data": { - "round_index": round_index, - "reflection": reflection, - }, - } + if emit_cot: + yield {"type": "round_finished", "data": trace.model_dump()} + yield { + "type": "reasoning_note", + "data": { + "round_index": round_index, + "reflection": reflection, + }, + } if len(all_usable) >= top_k or round_index >= max_rounds: break @@ -277,18 +287,19 @@ class AISearchAgent: ) # ── 7. Token-level LLM streaming ───────────────────────────────── - yield { - "type": "llm_started", - "data": {"page_count": len(pages_for_model), "result_count": len(results_for_model)}, - } - yield { - "type": "progress", - "data": { - "stage": "answering", - "message": "正在生成最终回答。", - "progress": 80, - }, - } + if emit_cot: + yield { + "type": "llm_started", + "data": {"page_count": len(pages_for_model), "result_count": len(results_for_model)}, + } + yield { + "type": "progress", + "data": { + "stage": "answering", + "message": "正在生成最终回答。", + "progress": 80, + }, + } answer: AnswerPayload | None = None async for chunk, final_answer in self.llm.answer_stream(SYSTEM_PROMPT, user_prompt, allowed_urls): @@ -306,10 +317,11 @@ class AISearchAgent: request.query, round_index + 1, latest_digest, len(all_usable) ) if extra_q and extra_q not in seen_q: - yield { - "type": "confidence_retry", - "data": {"confidence": answer.confidence, "extra_query": extra_q}, - } + if emit_cot: + yield { + "type": "confidence_retry", + "data": {"confidence": answer.confidence, "extra_query": extra_q}, + } try: _, extra_results = await self.search.search( extra_q, limit=self.MAX_SEARCH_HITS @@ -370,15 +382,16 @@ class AISearchAgent: } ) - yield { - "type": "final_meta", - "data": { - "search_mode": mode, - "timed_out": False, - "used_rounds": len(round_traces), - "used_hits": len(results_for_model), - }, - } + if emit_cot: + yield { + "type": "final_meta", + "data": { + "search_mode": mode, + "timed_out": False, + "used_rounds": len(round_traces), + "used_hits": len(results_for_model), + }, + } yield {"type": "result", "data": response.model_dump()} # ------------------------------------------------------------------ # diff --git a/ai_search_agent/api.py b/ai_search_agent/api.py index e7629a4..ac77b9c 100644 --- a/ai_search_agent/api.py +++ b/ai_search_agent/api.py @@ -2,7 +2,6 @@ from __future__ import annotations import asyncio import json -import os from fastapi import FastAPI from fastapi.responses import JSONResponse @@ -26,36 +25,25 @@ def create_app(enable_streaming: bool = True) -> FastAPI: @app.post("/search") async def search(request: SearchRequest) -> JSONResponse: - mode = agent.resolve_search_mode(request) - timeout_seconds = agent.get_time_budget_seconds(request) - try: - response = await asyncio.wait_for(agent.run(request), timeout=timeout_seconds) - except asyncio.TimeoutError: - return JSONResponse( - status_code=504, - content={ - "query": request.query, - "search_mode": mode, - "error": f"TIMEOUT_{mode.upper()}_SEARCH", - "message": f"{mode} search exceeded {timeout_seconds}s time budget", - }, - ) - audit_id = audit_logger.write(request, response) - response = response.model_copy(update={"audit_id": audit_id}) - return JSONResponse(content=response.model_dump()) + return JSONResponse( + status_code=410, + content={ + "error": "SEARCH_ENDPOINT_REMOVED", + "message": "Use /search/stream with search_mode=fast|deep.", + "query": request.query, + "search_mode": agent.resolve_search_mode(request), + }, + ) @app.post("/search/stream") async def search_stream(request: SearchRequest): if not enable_streaming: - stream_base_url = os.getenv("STREAM_SERVICE_BASE_URL", "").rstrip("/") return JSONResponse( status_code=501, content={ "error": "STREAMING_NOT_AVAILABLE_IN_FUNCTIONS", "message": "Use the always-on stream service for /search/stream.", - "stream_endpoint": ( - f"{stream_base_url}/search/stream" if stream_base_url else None - ), + "stream_endpoint": None, "query": request.query, "search_mode": agent.resolve_search_mode(request), }, diff --git a/function_app.py b/function_app.py index b348c4e..a89f6eb 100644 --- a/function_app.py +++ b/function_app.py @@ -6,9 +6,7 @@ from starlette.routing import Mount from ai_search_agent.api import create_app -# Azure Functions keeps only synchronous search endpoints. -# Stream endpoint should be served by always-on uvicorn service. -_fastapi = create_app(enable_streaming=False) +_fastapi = create_app() # Azure Functions delivers the full path including /api/ prefix to the ASGI app. # Mount FastAPI under /api so that /api/health -> FastAPI /health, etc.