""" Jina DeepSearch 流式集成测试 分三步： 1. 测试 SSE 原始流（验证 API 连通性） 2. 测试 jina_search / jina_reader 工具 3. 测试 jina_deepsearch @tool（完整流式采集） """ import json import os import re import sys import time import requests JINA_API_KEY = os.environ["JINA_API_KEY"] SEP = "=" * 60 # ───────────────────────────────────────────── # Step 1: jina_search # ───────────────────────────────────────────── def test_search(): print(f"\n{SEP}") print("STEP 1 — jina_search (s.jina.ai)") print(SEP) t0 = time.time() resp = requests.post( "https://s.jina.ai/", headers={ "Authorization": f"Bearer {JINA_API_KEY}", "Accept": "application/json", "X-Return-Format": "markdown", }, json={"q": "Jina AI DeepSearch API streaming 2025", "num": 2}, timeout=30, ) resp.raise_for_status() data = resp.json() items = data.get("data", []) print(f"✓ {len(items)} results in {time.time()-t0:.1f}s") for i, it in enumerate(items, 1): print(f" [{i}] {it.get('title','?')} — {it.get('url','?')[:60]}") content = (it.get("content") or "")[:200].replace("\n", " ") print(f" {content}…") return True # ───────────────────────────────────────────── # Step 2: jina_reader # ───────────────────────────────────────────── def test_reader(): print(f"\n{SEP}") print("STEP 2 — jina_reader (r.jina.ai)") print(SEP) url = "https://jina.ai/deepsearch" t0 = time.time() resp = requests.get( f"https://r.jina.ai/{url}", headers={ "Authorization": f"Bearer {JINA_API_KEY}", "Accept": "application/json", "X-Return-Format": "markdown", }, timeout=30, ) resp.raise_for_status() payload = resp.json().get("data", {}) title = payload.get("title", url) content = (payload.get("content") or "").strip() print(f"✓ Read '{title}' in {time.time()-t0:.1f}s") print(f" Content length: {len(content)} chars") print(f" Preview: {content[:300].replace(chr(10),' ')}…") return True # ───────────────────────────────────────────── # Step 3: jina_deepsearch — 原始 SSE 流 # ───────────────────────────────────────────── def test_deepsearch_stream(): print(f"\n{SEP}") print("STEP 3 — jina_deepsearch SSE streaming") print(SEP) query = "What is Jina AI DeepSearch and how does it differ from RAG? Brief answer." print(f"Query: {query}\n") headers = { "Authorization": f"Bearer {JINA_API_KEY}", "Content-Type": "application/json", "Accept": "text/event-stream", } payload = { "model": "jina-deepsearch-v1", "messages": [{"role": "user", "content": query}], "stream": True, "reasoning_effort": "low", "no_direct_answer": True, } full_content = "" citations = [] chunk_count = 0 think_chars = 0 answer_chars = 0 in_think = False t0 = time.time() print("── Streaming chunks ──") with requests.post( "https://deepsearch.jina.ai/v1/chat/completions", headers=headers, json=payload, stream=True, timeout=180, ) as resp: resp.raise_for_status() print(f"HTTP {resp.status_code} — headers: {dict(resp.headers).get('Content-Type','?')}") for raw_line in resp.iter_lines(): if not raw_line: continue line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line if not line.startswith("data: "): continue data_str = line[6:].strip() if data_str == "[DONE]": print("\n── [DONE] ──") break try: chunk = json.loads(data_str) except json.JSONDecodeError: continue chunk_count += 1 choice = (chunk.get("choices") or [{}])[0] delta = choice.get("delta", {}) piece = delta.get("content") or "" if piece: full_content += piece # track think vs answer for ch in piece: if full_content.endswith(""): in_think = True elif full_content.endswith(""): in_think = False if in_think: think_chars += len(piece) sys.stdout.write("·") # thinking indicator else: answer_chars += len(piece) sys.stdout.write("█") # answer indicator sys.stdout.flush() for ann in delta.get("annotations") or []: if ann.get("type") == "url_citation": citations.append(ann.get("url_citation", {})) elapsed = time.time() - t0 print(f"\n\n── Stats ──") print(f" Chunks received : {chunk_count}") print(f" Elapsed : {elapsed:.1f}s") print(f" Think chars : {think_chars}") print(f" Answer chars : {answer_chars}") print(f" Citations found : {len(citations)}") # Post-process answer = re.sub(r".*?", "", full_content, flags=re.DOTALL).strip() print(f"\n── Final Answer (first 800 chars) ──") print(answer[:800]) if citations: print(f"\n── Sources ({len(citations)}) ──") seen = set() i = 1 for c in citations: url = c.get("url", "") if url not in seen: seen.add(url) print(f" [{i}] {c.get('title','?')[:60]} — {url[:70]}") i += 1 return answer # ───────────────────────────────────────────── # Step 4: 通过 @tool 调用（验证 LangChain 兼容） # ───────────────────────────────────────────── def test_tool_invoke(): print(f"\n{SEP}") print("STEP 4 — jina_deepsearch via @tool .invoke()") print(SEP) # import from our module sys.path.insert(0, "/home/yzq/tmp/tech/deepagents") from jina_tools import jina_search, jina_deepsearch print("Testing jina_search.invoke()…") t0 = time.time() result = jina_search.invoke({"query": "Jina AI 2025 新功能", "num_results": 2}) print(f"✓ {time.time()-t0:.1f}s — {len(result)} chars returned") print(result[:300]) print("\nTesting jina_deepsearch.invoke()…") t0 = time.time() result = jina_deepsearch.invoke({ "query": "Jina Reader API vs FireCrawl comparison 2025", "reasoning_effort": "low", }) print(f"✓ {time.time()-t0:.1f}s — {len(result)} chars returned") print(result[:600]) # ───────────────────────────────────────────── # Main # ───────────────────────────────────────────── if __name__ == "__main__": try: test_search() except Exception as e: print(f"✗ search failed: {e}") try: test_reader() except Exception as e: print(f"✗ reader failed: {e}") try: test_deepsearch_stream() except Exception as e: print(f"✗ deepsearch failed: {e}") try: test_tool_invoke() except Exception as e: print(f"✗ tool invoke failed: {e}") print(f"\n{SEP}") print("All tests complete.") print(SEP)