Gemini 4 Argon 接入 飞书 完整指南

1✅ 群里贴的报错截图 / 监控图解读
2✅ Excel / CSV 表格的异常检测与透视结论
3✅ 白板照片、手写会议记录整理成结构化纪要
4✅ 长 PDF 合同 / 招标文件的条款问答
5✅ 产品原型图 → 前端页面结构的初稿
6✅ 多语言（尤其英日韩）邮件的中文改写与再创作

1❌ 纯短文本闲聊和 FAQ（用轻量型号便宜十倍）
2❌ 对中文术语精准度要求极高的政企公文（国产模型更贴地）
3❌ 需要秒级响应的定时任务推送

1import google.generativeai as genai
2import os, base64
3
4genai.configure(
5    api_key=os.environ["GEMINI_API_KEY"],
6    # client_options 里可以指定中转网关域名：https://你的网关域名
7    # transport="rest",
8)
9
10model = genai.GenerativeModel(
11    model_name="gemini-4-argon",
12    system_instruction="你是飞书群里的表格分析助手。回答用中文，先给结论，再给依据行号，不要客套话。",
13    generation_config={
14        "temperature": 0.2,
15        "maxOutputTokens": 8192,
16        "thinkingConfig": {"thinkingBudget": 2048},   # 思考预算：复杂表格调大，简单任务设 0 省钱降延迟
17    },
18)
19
20# 纯文本
21resp = model.generate_content("下面这张表哪一列有异常？" + csv_text)
22
23# 带图：image_bytes 是飞书下载回来的二进制
24resp = model.generate_content([
25    "这张图里的报错是什么意思，按可能性排三条例原因",
26    {"mime_type": "image/png", "data": base64.b64encode(image_bytes).decode()},
27])
28print(resp.text)

1from openai import OpenAI
2import base64
3
4client = OpenAI(
5    api_key=os.environ["GEMINI_API_KEY"],
6    # ❗ 关键：必须带 /v1beta/openai 后缀，少了这一段就是 404
7    base_url="https://generativelanguage.googleapis.com/v1beta/openai",
8)
9
10resp = client.chat.completions.create(
11    model="gemini-4-argon",
12    messages=[
13        {"role": "system", "content": "你是飞书群里的表格分析助手，回答用中文，200 字以内。"},
14        {
15            "role": "user",
16            "content": [
17                {"type": "text", "text": "这张图里的报错是什么意思？"},
18                {
19                    "type": "image_url",
20                    "image_url": {"url": "data:image/png;base64," + base64.b64encode(image_bytes).decode()},
21                },
22            ],
23        },
24    ],
25    max_tokens=4096,
26)
27print(resp.choices[0].message.content)

1待签字符串 = timestamp + "\n" + secret
2签名 = HMAC-SHA256(key = secret, message = 待签字符串)，取小写十六进制 hex

1import hashlib, hmac, json, os, time
2import requests
3
4WEBHOOK = os.environ["FEISHU_WEBHOOK_URL"]
5SECRET  = os.environ["FEISHU_BOT_SECRET"]
6
7def gen_sign(timestamp: int, secret: str) -> str:
8    string_to_sign = f"{timestamp}\n{secret}"
9    return hmac.new(
10        secret.encode("utf-8"),
11        string_to_sign.encode("utf-8"),
12        hashlib.sha256,
13    ).hexdigest()          # ❗ hex，不是 base64
14
15def push(body: dict) -> dict:
16    ts = int(time.time())                 # ❗ 秒，不是毫秒
17    payload = dict(body, timestamp=str(ts), sign=gen_sign(ts, SECRET))
18    r = requests.post(
19        WEBHOOK,
20        headers={"Content-Type": "application/json; charset=utf-8"},
21        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
22        timeout=10,
23    )
24    return r.json()        # 必须判断 Code == 0，光看 HTTP 200 会被坑

1import google.generativeai as genai
2
3genai.configure(api_key=os.environ["GEMINI_API_KEY"])
4model = genai.GenerativeModel("gemini-4-argon")
5
6def build_report(csv_text: str) -> str:
7    prompt = (
8        "下面是今天的核心业务数据表（CSV）。请输出：\n"
9        "1. 三句话讲清今天最值得关注的变化\n"
10        "2. 异常波动的指标，列出值和环比\n"
11        "3. 一条可执行建议\n"
12        "总字数 220 字以内，不要标题不要客套话。\n\n" + csv_text
13    )
14    return model.generate_content(prompt).text
15
16push({"msg_type": "text", "content": {"text": build_report(csv_text)}})

1{ "Extra": null, "StatusCode": 0, "StatusMessage": "success" }

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": {
5      "wide_screen_mode": true,
6      "enable_forward": true
7    },
8    "header": {
9      "template": "turquoise",
10      "title": { "tag": "plain_text", "content": "每日数据日报 · 2026-06-29" }
11    },
12    "elements": [
13      {
14        "tag": "div",
15        "text": { "tag": "lark_md", "content": "**GMV**  ¥ 1,284,300（↑ 6.2%）\n**订单量**  8,431（↑ 2.1%）\n**退款率**  3.8%（↑ 0.9pp ⚠️）" }
16      },
17      { "tag": "hr" },
18      { "tag": "markdown", "content": "Gemini 4 Argon 结论：退款率抬头集中在 SKU-A7 批次，建议今日人工复核该批次发货记录。" },
19      {
20        "tag": "img",
21        "alt": { "tag": "plain_text", "content": "GMV 趋势" },
22        "img_key": "img_v3_xxxxxxxx"            // ❗ 必须先用图片上传接口拿到 key
23      },
24      {
25        "tag": "action",
26        "actions": [
27          {
28            "tag": "button",
29            "text": { "tag": "plain_text", "content": "打开看板" },
30            "type": "primary",
31            "url": "https://bi.example.com/daily"
32          }
33        ]
34      },
35      {
36        "tag": "note",
37        "elements": [{ "tag": "plain_text", "content": "自动生成，数据口径以 BI 系统为准" }]
38      }
39    ]
40  }
41}

1App ID:     cli_a1b2c3d4e5f6g7h8
2App Secret: Xd8xxxxxxxxxxxxxxxxxxxxxxxxxxxx

1import threading, time, requests
2
3TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
4
5class TokenCache:
6    """有效期 7200 秒；多实例请换成 Redis 共享，别每台机器自己一套"""
7    def __init__(self, app_id, app_secret):
8        self.app_id, self.app_secret = app_id, app_secret
9        self._token, self._expire_at, self._lock = None, 0, threading.Lock()
10
11    def get(self):
12        with self._lock:
13            if time.time() < self._expire_at:
14                return self._token
15            r = requests.post(TOKEN_URL, json={
16                "app_id": self.app_id, "app_secret": self.app_secret,
17            }, timeout=10).json()
18            if r.get("code") != 0:
19                raise RuntimeError(f"token 获取失败: {r}")
20            self._token = r["tenant_access_token"]
21            self._expire_at = time.time() + r.get("expire", 7200) - 300   # 提前 5 分钟续
22            return self._token
23
24tokens = TokenCache(os.environ["FEISHU_APP_ID"], os.environ["FEISHU_APP_SECRET"])

1def download_image(message_id: str, image_key: str) -> bytes:
2    """❗ 必须带 Authorization，匿名拉会 403；image_key 也有有效期，别长期缓存"""
3    url = (f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}"
4           f"/resources/{image_key}?type=image")
5    r = requests.get(url, headers={"Authorization": "Bearer " + tokens.get()}, timeout=20)
6    if r.status_code != 200 or not r.content:
7        raise RuntimeError(f"图片下载失败: {r.status_code}")
8    return r.content

1def send_message(receive_id: str, content: dict, msg_type: str = "text", id_type: str = "chat_id"):
2    url = f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type={id_type}"
3    # ❗ content 一律是「JSON 字符串」，不是对象；msg_type 可为 text / post / interactive
4    body = {
5        "receive_id": receive_id,
6        "msg_type": msg_type,
7        "content": json.dumps(content, ensure_ascii=False),
8    }
9    r = requests.post(url, headers={"Authorization": "Bearer " + tokens.get()},
10                      json=body, timeout=15).json()
11    if r.get("code") != 0:
12        raise RuntimeError(f"发送失败 {r.get('code')}: {r.get('msg')}")
13    return r["data"]["message_id"]

1import json
2import lark_oapi as lark
3
4def on_message(data: lark.im.v1.P2ImMessageReceiveV1) -> None:
5    msg = data.event.message
6    content = json.loads(msg.content)
7
8    if msg.message_type == "image":
9        img = download_image(msg.message_id, content["image_key"])
10        answer = ask_gemini_with_image("这张图讲了什么？按重要性给三条结论", img)
11    else:
12        answer = ask_gemini_text(content.get("text", "").strip())
13
14    if answer:
15        send_message(msg.chat_id, {"text": answer})
16
17ws = lark.ws.Client(
18    os.environ["FEISHU_APP_ID"],
19    os.environ["FEISHU_APP_SECRET"],
20    event_handler=lark.EventDispatcherHandler.builder("", "")
21        .register_p2_im_message_receive_v1(on_message)
22        .build(),
23    log_level=lark.LogLevel.INFO,
24)
25ws.start()

1from Crypto.Cipher import AES
2import base64, hashlib, json
3
4def decrypt(encrypted_b64: str, encrypt_key: str) -> dict:
5    key = hashlib.sha256(encrypt_key.encode()).digest()   # 32 字节
6    raw = base64.b64decode(encrypted_b64)
7    cipher = AES.new(key, AES.MODE_CBC, raw[:16])          # 前 16 字节是 IV
8    plain = cipher.decrypt(raw[16:])
9    return json.loads(plain[: -plain[-1]].decode("utf-8")) # 去 PKCS7 填充

1这是我们生产环境的报错截图。请完成：
2
31. 识别图中错误信息的核心关键词
42. 按可能性排三条根因假设，每条一句
53. 每条给出一条可以直接执行的验证命令或 SQL
64. 如果图里有堆栈，指出最关键的那一行并说明理由
7
8背景：Java + Spring Boot 3、MySQL 8、部署在 K8s 上，昨天下午做过一次发版。

1下面是本周的用户行为明细表 CSV（约 2000 行）。请输出：
2
31. 三个最值得注意的异常，每条写清楚「哪个维度 × 哪个值 × 偏离多少」
42. 给出你认为最有解释力的一个原因，并说明你会用什么数据验证它
53. 一句话给下周的运营建议
6
7要求：不要复述表格全部内容，不要写「综上所述」，直接给结论。表格：
8<粘贴 CSV>

1这是我们刚才白板讨论的照片，字迹较潦草。请：
2
31. 逐条识别白板上的要点，识别不出来的地方标注「[模糊]」，绝不允许猜测补全
42. 把识别出的要点按「问题 / 方案 / 待办」三类归类
53. 输出一张 Markdown 表格，表头为 分类 | 内容 | 是否达成共识
64. 最后用三句话总结这次讨论的核心分歧