HuiBrain/docs/IMAGE_API_GUIDE.md

EduBrain 图片功能集成指南

本文档介绍如何在 ADK Agent 项目中集成 EduBrain 的图片 OCR 和搜索功能。

服务地址

BASE_URL = http://localhost:8765

所有接口前缀：/api/v1/images

---

一、接口总览

方法	路径	说明	请求类型
POST	/recognize-direct	上传并识别单张图片	multipart/form-data
POST	/batch-recognize	批量上传并识别	multipart/form-data
POST	/import-paths	从服务器路径导入	JSON
GET	/search	AI 搜索图片（SSE 流式）	query params
GET	/{image_id}	获取单张图片详情	path param
GET	``	获取图片列表（分页）	query params

---

二、上传并识别图片

接口

POST /api/v1/images/recognize-direct
Content-Type: multipart/form-data

参数

字段	类型	必填	说明
file	File	是	图片文件，支持 .png .jpg .jpeg .bmp .webp

返回

{
    "id": 386,
    "file_path": "data/images/test.png",
    "original_filename": "test.png",
    "ocr_text": "识别出的文字内容...",
    "tags": [],
    "confidence": 1.0,
    "provider": "deepseek",
    "status": "completed",
    "blocks": [
        {"text": "第一段文字", "bbox": [x1,y1,x2,y2], "confidence": 0.98}
    ]
}

Python 调用示例

import httpx

async def recognize_image(file_path: str) -> dict:
    async with httpx.AsyncClient(timeout=120.0) as client:
        with open(file_path, "rb") as f:
            resp = await client.post(
                f"{BASE_URL}/api/v1/images/recognize-direct",
                files={"file": (file_path, f)}
            )
        return resp.json()

---

三、批量上传并识别

接口

POST /api/v1/images/batch-recognize
Content-Type: multipart/form-data

参数

字段	类型	必填	说明
files	File[]	是	多个图片文件

返回

{
    "total": 3,
    "success": 2,
    "failed": 1,
    "results": [
        {
            "id": 387,
            "filename": "img1.png",
            "file_path": "data/images/img1.png",
            "status": "completed",
            "ocr_text": "识别内容...",
            "tags": [],
            "confidence": 1.0,
            "provider": "deepseek"
        },
        {
            "id": 388,
            "filename": "img2.jpg",
            "file_path": "data/images/img2.jpg",
            "status": "failed",
            "message": "OCR 识别失败: ..."
        }
    ]
}

---

四、从服务器路径导入

接口

POST /api/v1/images/import-paths
Content-Type: application/json

请求体

{
    "paths": ["/data/images/001.png", "/data/images/002.jpg"],
    "recursive": false
}

返回

与批量上传相同格式。

---

五、AI 搜索图片（SSE 流式）⚠️ 重点

接口

GET /api/v1/images/search?keyword=搜索词&limit=5&sort=time_desc

这不是普通 HTTP 请求，而是 SSE（Server-Sent Events）流式接口。

参数

参数	类型	必填	默认值	说明
keyword	string	是	-	搜索关键词
limit	int	否	5	最多返回多少条匹配结果
sort	string	否	time_desc	排序方式：time_desc（时间倒序）、time_asc（时间正序）、random（随机）

搜索原理

用户搜索 "孩子不想上学"
  → 从数据库取所有已识别图片（按 sort 排序）
  → 每批 10 张发给 MiniMax M2.7 判断是否匹配
  → 并发池最多 10 个 LLM 请求同时执行
  → 匹配结果通过 SSE 实时推送（找到就立刻返回）
  → 找够 limit 条或遍历完所有图片后结束

SSE 事件格式

连接后，服务端会持续推送 JSON 事件，每条格式为：

data: {"type": "事件类型", ...其他字段}\n\n

事件类型

1. start - 搜索开始

{"type": "start", "keyword": "孩子不想上学", "limit": 5}

2. progress - 进度更新

{"type": "progress", "checked": 170, "total": 385, "found": 3}

字段	类型	说明
checked	int	已检查的图片数量
total	int	数据库中图片总数
found	int	已找到的匹配数量

3. result - 匹配结果（实时推送）

{
    "type": "result",
    "id": 42,
    "file_path": "data/images/001.png",
    "image_url": "001.png",
    "image_base64": "/9j/4QE3RXhpZgAATU0AKgAAAAgABgEAAAQ...",
    "ocr_text": "完整的 OCR 识别文本...",
    "ocr_text_preview": "前150个字符的预览...",
    "tags": ["亲子关系", "沟通"],
    "confidence": 1.0,
    "provider": "deepseek",
    "created_at": "2026-04-11T21:59:56"
}

字段	类型	说明
id	int	图片记录 ID
file_path	string	图片文件路径（服务器本地路径）
image_url	string	图片文件名（访问地址为 BASE_URL/data/images/{image_url}）
image_base64	string	图片的 Base64 编码，可直接用于发送到企业微信等
ocr_text	string	完整 OCR 文本
ocr_text_preview	string	OCR 文本预览（前150字符）
tags	string[]	关键词标签
confidence	float	OCR 置信度（0-1）
provider	string	OCR 提供商
created_at	string	创建时间（ISO 8601）

4. done - 搜索结束

{"type": "done", "total_found": 5, "total_checked": 385}

字段	类型	说明
total_found	int	总共找到的匹配数量
total_checked	int	总共检查的图片数量

Python 调用示例（推荐）

import httpx
import json
from typing import AsyncGenerator

BASE_URL = "http://localhost:8765"

async def search_images(
    keyword: str,
    limit: int = 5,
    sort: str = "time_desc",
) -> AsyncGenerator[dict, None]:
    """
    AI 搜索图片（SSE 流式）
    
    Yields:
        dict: 每个 SSE 事件解析后的字典
    """
    async with httpx.AsyncClient(timeout=300.0) as client:
        async with client.stream(
            "GET",
            f"{BASE_URL}/api/v1/images/search",
            params={"keyword": keyword, "limit": limit, "sort": sort},
        ) as resp:
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                data = json.loads(line[6:])
                yield data


# ── 使用示例 ──

async def demo_search():
    results = []
    
    async for event in search_images("孩子不想上学", limit=5):
        event_type = event.get("type")
        
        if event_type == "start":
            print(f"开始搜索: {event['keyword']}")
            
        elif event_type == "progress":
            print(f"进度: {event['checked']}/{event['total']}，已找到 {event['found']} 条")
            
        elif event_type == "result":
            results.append(event)
            print(f"找到匹配: [{event['id']}] {event['ocr_text_preview'][:50]}")
            
        elif event_type == "done":
            print(f"搜索完成: 共找到 {event['total_found']} 条，检查了 {event['total_checked']} 张")
    
    return results

ADK Agent 集成示例

在 ADK Agent 中，可以将搜索功能封装为一个 Tool：

from google.adk import Tool

class ImageSearchTool(Tool):
    """搜索图片知识库"""
    
    name = "search_images"
    description = "搜索已识别的图片库，返回与查询语义相关的图片及其 Base64 数据。支持自然语言查询。"
    
    async def execute(self, keyword: str, limit: int = 5) -> str:
        results = []
        async for event in search_images(keyword, limit=limit):
            if event["type"] == "result":
                results.append({
                    "id": event["id"],
                    "image_base64": event["image_base64"],
                    "image_url": event["image_url"],
                    "ocr_text_preview": event["ocr_text_preview"],
                    "tags": event.get("tags", []),
                })
            elif event["type"] == "done":
                pass  # 搜索结束
        
        if not results:
            return f"未找到与 '{keyword}' 相关的图片"
        
        # 返回 JSON，方便 Agent 后续处理（如发送到企业微信）
        return json.dumps({
            "keyword": keyword,
            "total": len(results),
            "images": results,
        }, ensure_ascii=False)

提示：image_base64 字段包含图片的完整 Base64 编码，可直接用于企业微信发送图片接口。 企业微信发送图片需要先通过「上传临时素材」接口上传 base64 获取 media_id，再发送图片消息。

---

六、获取图片详情

接口

GET /api/v1/images/{image_id}

返回

{
    "id": 1,
    "file_path": "data/images/001.png",
    "ocr_text": "完整 OCR 文本...",
    "confidence": 1.0,
    "provider": "deepseek",
    "status": "completed",
    "tags": null,
    "story_summary": null,
    "created_at": "2026-04-11T20:00:00",
    "updated_at": "2026-04-11T20:00:05"
}

---

七、获取图片列表

接口

GET /api/v1/images?page=1&page_size=20&status=completed

参数

参数	类型	必填	默认值	说明
page	int	否	1	页码
page_size	int	否	20	每页数量
status	string	否	-	筛选状态：completed、pending、processing、failed

返回

{
    "total": 385,
    "page": 1,
    "page_size": 20,
    "items": [
        {
            "id": 385,
            "file_path": "data/images/001.png",
            "ocr_text": "...",
            "confidence": 1.0,
            "provider": "deepseek",
            "status": "completed",
            "tags": null,
            "story_summary": null,
            "created_at": "2026-04-11T21:59:56",
            "updated_at": "2026-04-11T21:59:59"
        }
    ]
}

---

八、图片文件访问

识别后的图片可通过静态文件路径访问：

GET /data/images/{文件名}

例如：

http://localhost:8765/data/images/001.png

---

九、注意事项

1. 搜索接口是 SSE 流式，不能用普通的 requests.get() 或 httpx.get() 直接获取 JSON，必须用流式读取（stream() / iter_lines()）
2. 搜索耗时较长：385 张图片全部遍历约需 2-5 分钟，建议设置较长的超时（300 秒）
3. 搜索结果实时推送：不需要等所有图片检查完毕，匹配结果会逐条推送
4. limit 参数控制返回数量：找到 limit 条匹配后自动停止搜索
5. OCR 识别耗时：单张图片约 5-15 秒，建议设置 120 秒超时
6. 文件格式：仅支持 .png .jpg .jpeg .bmp .webp