DeepAgents × Jina DeepSearch
SSE 流式集成全流程报告

thinking_done = False
think_buf = ""

for raw_line in resp.iter_lines():
    if not raw_line.startswith("data: "): continue
    data_str = raw_line[6:].strip()
    if data_str == "[DONE]": break

    chunk = json.loads(data_str)
    delta = chunk["choices"][0]["delta"]
    piece = delta.get("content") or ""

    if piece:
        full_content += piece
        if not thinking_done:
            think_buf += piece
            if "</think>" in think_buf:
                thinking_done = True
                after = think_buf.split("</think>", 1)[1]
                if after: stream_or_print(after)
        else:
            stream_or_print(piece)  # ← 实时输出

    # citations：检查 delta + chunk 两层
    for ann in (delta.get("annotations") or chunk.get("annotations") or []):
        if ann.get("type") == "url_citation":
            citations.append(ann.get("url_citation") or ann)

for chunk, metadata in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="messages",   # ← 正确模式，返回 (AIMessageChunk, metadata)
):
    node    = metadata.get("langgraph_node", "")
    content = getattr(chunk, "content", "") or ""
    if content:
        print(content, end="", flush=True)

from langgraph.config import get_stream_writer

@tool
async def jina_deepsearch_stream(query: str, reasoning_effort: str = "low") -> str:
    write = get_stream_writer()

    async with httpx.AsyncClient(timeout=300) as client:
        async with client.stream("POST", _DEEPSEARCH_URL, ...) as resp:
            async for raw_line in resp.aiter_lines():
                ...
                if thinking_done:
                    write(piece)   # ← 每个 token 推送到 custom stream

# 调用端
async for mode, data in agent.astream(
    {"messages": [...]},
    stream_mode=["custom", "messages"],
):
    if mode == "custom":
        print(data, end="", flush=True)  # ← 理想：实时 token

@tool
async def streaming_counter(n: int = 5) -> str:
    write = get_stream_writer()
    for i in range(n):
        write(f"token_{i} ")
        await asyncio.sleep(0.1)
    return "done"

# 结果
custom events: 0   ← deepagents 工具执行层不透传 custom stream

import asyncio, json, os, re, sys
from typing import Annotated
import httpx
from langchain_core.tools import tool
from langgraph.config import get_stream_writer

@tool
async def jina_deepsearch_stream(
    query: Annotated[str, "Research question"],
    reasoning_effort: Annotated[str, "low/medium/high"] = "low",
) -> str:
    write = get_stream_writer()
    thinking_done = False
    think_buf = ""
    full_content = ""
    citations = []

    async with httpx.AsyncClient(timeout=300) as client:
        async with client.stream("POST", _DEEPSEARCH_URL,
            headers={
                "Authorization": f"Bearer {JINA_API_KEY}",
                "Accept": "text/event-stream",
                "Content-Type": "application/json",
            },
            json={
                "model": "jina-deepsearch-v1",
                "messages": [{"role": "user", "content": query}],
                "stream": True,
                "reasoning_effort": reasoning_effort,
                "no_direct_answer": True,
            }) as resp:
            resp.raise_for_status()
            async for raw_line in resp.aiter_lines():
                if not raw_line.startswith("data: "): continue
                data_str = raw_line[6:].strip()
                if data_str == "[DONE]": break
                chunk = json.loads(data_str)
                delta = (chunk.get("choices") or [{}])[0].get("delta", {})
                piece = delta.get("content") or ""
                if piece:
                    full_content += piece
                    if not thinking_done:
                        think_buf += piece
                        if "</think>" in think_buf:
                            thinking_done = True
                            after = think_buf.split("</think>", 1)[1]
                            if after:
                                sys.stdout.write(after); sys.stdout.flush()
                                write(after)
                    else:
                        sys.stdout.write(piece); sys.stdout.flush()
                        write(piece)
                # collect citations
                for ann in (delta.get("annotations") or chunk.get("annotations") or []):
                    if isinstance(ann, dict) and ann.get("type") == "url_citation":
                        uc = ann.get("url_citation") or ann
                        if uc.get("url"): citations.append(uc)

    answer = re.sub(r"<think>.*?</think>", "", full_content, flags=re.DOTALL).strip()
    for i, c in enumerate(_dedupe_citations(citations), 1):
        answer += f"\n[{i}] {c.get('title','')} — {c.get('url','')}"
    return answer

import asyncio, os, sys, time
from langchain_openai import ChatOpenAI
from deepagents import create_deep_agent
from jina_tools import jina_deepsearch_stream, jina_search, jina_reader

model = ChatOpenAI(
    model="qwen3.5-35b-a3b",
    base_url="https://openrouter.yangzhiqiang.tech/api/v1",
    api_key="sk-123",
    temperature=0, streaming=True,
)

agent = create_deep_agent(
    model=model,
    tools=[jina_deepsearch_stream, jina_search, jina_reader],  # 直挂主 agent
    system_prompt="Use jina_deepsearch_stream to research then write the article.",
    subagents=[],  # 无 subagent，避免 custom stream 被阻断
)

async def main():
    async for mode, data in agent.astream(
        {"messages": [{"role": "user", "content": QUERY}]},
        stream_mode=["custom", "messages"],
    ):
        if mode == "custom":
            print(data, end="", flush=True)  # 批量到达，非逐 token
        elif mode == "messages":
            chunk, meta = data
            if meta.get("langgraph_node") == "model":
                print(getattr(chunk, "content", ""), end="", flush=True)

asyncio.run(main())