输入一个问题，调用 Felo 搜索 API（POST /v2/chat）

输入一个问题，调用 Felo 搜索 API（POST /v2/chat），得到 AI 回答、Felo 实际执行的搜索查询（query_analysis）以及来源列表（resources），整理成一份带检索时间和来源链接的调研简报。遇到 429 按 Retry-After 等待后重试；以实测的错误格式为准统一处理错误。

先看清成品

• 测试： pytest 5 项通过，覆盖正常结果、无效 Key、429 重试、没有来源时的提示、query 长度限制。
• 模拟服务器： 生成了一份 3 个来源的简报。
• 真实接口： 用无效 Key 调用，程序输出“✗ Felo 返回错误：HTTP 401 UNAUTHORIZED Invalid API Key”。

没有在 box 上跑的部分： box 上没有 Felo API Key。对真实接口只做过无效 Key 的调用：/v2/chat 和 /api/v1/models 都返回 HTTP 401，body 是 {"status":401,"code":"UNAUTHORIZED","message":"Invalid API Key",…}；不带 Key 时 message 是 "API Key is required"。其余部分在我写的模拟服务器 fake_felo.py 上运行，它按文档的结构返回固定的假内容。

合规提醒：FELO_API_KEY 只放在环境变量或服务器端，不要写进网页前端或提交到 Git。问题内容会发到 Felo 的服务器，不要提交个人隐私或公司机密。AI 回答引用网页时，版权归原作者：保留来源链接，引用前打开原文核对，不要整篇转载。检查链接时请求要少、要慢，并遵守对方网站的规则。计费规则以 Felo API Platform 的页面为准，本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

46-felo-A-src/
├── felo_search.py
├── fake_felo.py
└── test_felo.py

第 1 步：代码

felo_search.py：

"""Felo 搜索 API（POST https://openapi.felo.ai/v2/chat）→ 带来源的调研简报（Markdown）。
- 成功：status == "ok"，data 里有 answer、query_analysis.queries、resources（link/title/snippet）
- 失败：HTTP 状态码非 200，或 status != "ok"；401 的 body 形如 {"status":401,"code":"UNAUTHORIZED","message":"Invalid API Key"}（实测）
- 429：按 Retry-After 等待后重试（文档：/v2/chat 有按 key 和按用户的分钟、小时限流）
环境变量：FELO_API_KEY、FELO_BASE（默认 https://openapi.felo.ai）
用法：python felo_search.py "问题" [--out 简报.md]"""
import argparse, os, sys, time
from datetime import datetime
import requests

BASE = os.environ.get("FELO_BASE", "https://openapi.felo.ai")

class FeloError(Exception): pass

def ask(query, retries=3):
    if not 1 <= len(query) <= 2000: raise FeloError("query 长度必须在 1~2000 字符之间（文档限制）")
    key = os.environ.get("FELO_API_KEY") or sys.exit("请先设置环境变量 FELO_API_KEY")
    for i in range(retries + 1):
        r = requests.post(f"{BASE}/v2/chat", json={"query": query}, timeout=120,
                          headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"})
        if r.status_code == 429 and i < retries:
            wait = float(r.headers.get("Retry-After", 2 ** i)); print(f"  429 限流，{wait:.0f}s 后重试", file=sys.stderr)
            time.sleep(wait); continue
        try: body = r.json()
        except ValueError: raise FeloError(f"HTTP {r.status_code}，返回的不是 JSON：{r.text[:200]}")
        if r.status_code != 200 or body.get("status") != "ok":
            raise FeloError(f"HTTP {r.status_code} {body.get('code', '')} {body.get('message')}".strip())
        return body["data"]
    raise FeloError("多次重试后仍被限流")

def to_markdown(query, d):
    lines = [f"# 调研简报：{query}", "", f"- 检索时间：{datetime.now():%Y-%m-%d %H:%M}（来源：Felo 搜索 API）",
             f"- Felo 实际搜索的查询：{'；'.join(d.get('query_analysis', {}).get('queries', [])) or '无'}", "",
             "## 回答", "", d.get("answer", "").strip(), "", "## 来源", ""]
    res = d.get("resources") or []
    lines += [f"{i}. [{x.get('title') or x['link']}]({x['link']})：{x.get('snippet', '')}" for i, x in enumerate(res, 1)] or ["（没有返回来源，这个回答不要直接引用）"]
    lines += ["", "> AI 生成的回答，引用前请打开来源链接核对原文。"]
    return "\n".join(lines) + "\n"

def main():
    ap = argparse.ArgumentParser(); ap.add_argument("query"); ap.add_argument("--out")
    a = ap.parse_args()
    try: d = ask(a.query)
    except FeloError as e: sys.exit(f"✗ Felo 返回错误：{e}")
    md = to_markdown(a.query, d)
    if a.out: open(a.out, "w", encoding="utf-8").write(md); print(f"✓ 已写入 {a.out}（{len(d.get('resources') or [])} 个来源）")
    else: print(md)

if __name__ == "__main__":
    main()

fake_felo.py（仅用于离线测试）：

"""Felo API 的模拟服务器（仅用于离线测试，按 openapi.felo.ai 文档的请求/响应结构编写，内容是假的）。
- POST /v2/chat：key 不对返回 401 {"status":401,"code":"UNAUTHORIZED","message":"Invalid API Key"}（和真实接口实测一致）
  query 含“限流”时前 1 次返回 429 + Retry-After: 1
- GET /api/v1/models、POST /api/v1/chat/completions：OpenAI 兼容格式
- GET /page/<n>：示例网页；/page/404 返回 404
用法：python fake_felo.py   （监听 127.0.0.1:18460，测试 key 为 fk-test）"""
from flask import Flask, jsonify, request
app = Flask(__name__); hits = {"rl": 0}
BASE = "http://127.0.0.1:18460"
RES = [{"link": f"{BASE}/page/1", "title": "示例：电动牙刷怎么选", "snippet": "选择电动牙刷时，先看刷头更换成本和压力提醒。"},
       {"link": f"{BASE}/page/2", "title": "示例：声波和旋转式的区别", "snippet": "声波式振动频率高，旋转式清洁力度更集中。"},
       {"link": f"{BASE}/page/404", "title": "示例：一个已经失效的链接", "snippet": "这条链接会返回 404。"}]

def bad(): return request.headers.get("Authorization") != "Bearer fk-test"
def unauth(): return jsonify(status=401, code="UNAUTHORIZED", message="Invalid API Key"), 401

@app.post("/v2/chat")
def chat():
    if bad(): return unauth()
    q = (request.get_json(silent=True) or {}).get("query", "")
    if not 1 <= len(q) <= 2000: return jsonify(status="error", message="query must be 1-2000 characters"), 400
    if "限流" in q and hits["rl"] < 1:
        hits["rl"] += 1; return jsonify(status="error", message="rate limited"), 429, {"Retry-After": "1"}
    res = [] if "冷门" in q else RES
    return jsonify(status="ok", message=None, data={"id": "fake-session", "message_id": "fake-msg",
        "answer": "（模拟回答）" + ("没有找到资料。" if not res else "选电动牙刷先看刷头成本[1]，再看振动方式[2]。"),
        "query_analysis": {"queries": [q, q + " 推荐"]}, "resources": res})

@app.get("/api/v1/models")
def models():
    if bad(): return unauth()
    return jsonify(object="list", data=[{"id": "fake-small", "object": "model"}, {"id": "fake-large", "object": "model"}])

@app.post("/api/v1/chat/completions")
def completions():
    if bad(): return unauth()
    b = request.get_json(); m = b["model"]
    if m not in ("fake-small", "fake-large"): return jsonify(error={"message": f"model {m} not found"}), 404
    return jsonify(id="c1", object="chat.completion", created=0, model=m,
                   choices=[{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": f"（模拟）{m} 收到：{b['messages'][-1]['content'][:20]}"}}],
                   usage={"prompt_tokens": 10, "completion_tokens": 8, "total_tokens": 18})

@app.route("/page/<pid>", methods=["GET", "HEAD"])
def page(pid): return ("not found", 404) if pid == "404" else (f"示例页面 {pid}", 200)

if __name__ == "__main__":
    app.run(port=18460)

test_felo.py：

import os, subprocess, sys, time, pytest
os.environ.update(FELO_BASE="http://127.0.0.1:18460", FELO_API_KEY="fk-test")
import felo_search as f

@pytest.fixture(scope="module", autouse=True)
def server():
    p = subprocess.Popen([sys.executable, os.path.join(os.path.dirname(__file__), "fake_felo.py")], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    time.sleep(1.5); yield; p.terminate()

def test_answer_and_sources():
    md = f.to_markdown("电动牙刷", f.ask("电动牙刷怎么选"))
    assert "## 来源" in md and "/page/1" in md and "检索时间" in md

def test_bad_key_raises(monkeypatch):
    monkeypatch.setenv("FELO_API_KEY", "wrong")
    with pytest.raises(f.FeloError, match="401 UNAUTHORIZED"): f.ask("x")

def test_retry_after_429():
    assert f.ask("限流测试")["answer"]

def test_no_sources_warned():
    assert "没有返回来源" in f.to_markdown("q", f.ask("冷门问题"))

def test_query_length():
    with pytest.raises(f.FeloError): f.ask("")

第 2 步：运行

$ export FELO_API_KEY=...
$ python felo_search.py "电动牙刷怎么选" --out 简报.md
# box 上：python fake_felo.py 后设置 FELO_BASE=http://127.0.0.1:18460 FELO_API_KEY=fk-test
✓ 已写入 简报.md（3 个来源）

简报.md（模拟数据）：

# 调研简报：电动牙刷怎么选

- 检索时间：2026-10-21 17:15（来源：Felo 搜索 API）
- Felo 实际搜索的查询：电动牙刷怎么选；电动牙刷怎么选 推荐

## 回答

（模拟回答）选电动牙刷先看刷头成本[1]，再看振动方式[2]。

## 来源

1. [示例：电动牙刷怎么选](http://127.0.0.1:18460/page/1)：选择电动牙刷时，先看刷头更换成本和压力提醒。
2. [示例：声波和旋转式的区别](http://127.0.0.1:18460/page/2)：声波式振动频率高，旋转式清洁力度更集中。
3. [示例：一个已经失效的链接](http://127.0.0.1:18460/page/404)：这条链接会返回 404。

> AI 生成的回答，引用前请打开来源链接核对原文。

检查点：看一下“Felo 实际搜索的查询”这一行，它说明了搜索引擎是怎样理解你的问题的。如果理解偏了，就改写问题。

调试示例：文档写的是 "error"，实测返回的是 401

文档说明失败时 status 为 "error"。如果照着文档写成 if body["status"] == "error": 报错，用无效 Key 一测就会出问题。实测返回的是：

{"status":401,"code":"UNAUTHORIZED","message":"Invalid API Key","request_id":"…"}

这里 status 是整数 401，不是字符串 "error"。照文档写的判断不会生效，程序会接着去读 body["data"]，抛出 KeyError，看不出到底哪里错了。改成“HTTP 状态码不是 200，或者 status 不是 "ok”就报错，并把 code 和 message 带出来”之后，错误就一目了然了。教训：错误处理的分支要用真实返回测一次，无效 Key 就是成本最低的测试。

常见问题

模板规则

• 成功的判断要严格：只认 HTTP 200 加上 status "ok"。
• 简报必须带检索时间和来源。
• 错误分支用真实的无效 Key 测一次。

项目资源

一份带出处和检索时间的简报，才是能交给别人看的调研结果。