LangGraph Agent Runtime

将本地 Agent Skill 包装成 HTTP API 服务。用户发起 SSE 请求，Agent 自动选择 Skill、读取流程说明（SKILL.md）、调用脚本或搜索网页，最终流式返回结果。

核心设计原则

本系统遵循 “模型负责判断和编排，脚本负责确定性执行” 的架构：

• System Prompt 不硬编码业务逻辑，只引导 Agent 读取 SKILL.md
• SKILL.md 定义完整工作流，Agent 先读取再执行
• scripts/ 处理 API 调用、鉴权、分页等确定性逻辑
• Agent 负责意图理解、API 搜索发现、参数构建、多步数据收集

架构概览

调用方 (Java后端 / 小程序 / 浏览器)
        │
        │  GET /query/stream?question=...  (SSE)
        ▼
┌─────────────────────────────────┐
│  FastAPI                         │
│  ├── /healthz                   │
│  └── /query/stream              │
└──────────┬──────────────────────┘
           │
           ▼
┌─────────────────────────────────┐
│  LangGraph ReAct Agent          │
│                                 │
│  1. route_question → 选 Skill   │
│  2. 注入对应 System Prompt      │
│  3. 读取 SKILL.md 了解工作流   │
│  4. 用脚本搜索和调用 API        │
│  5. 组织最终回答                │
├─────────────────────────────────┤
│  tikhub-api-helper 工具:        │
│  ├── read_skill_file   读 SKILL │
│  └── run_skill_script  执行脚本 │
│                                 │
│  deep-research 工具:            │
│  ├── read_skill_file            │
│  ├── run_skill_script           │
│  ├── web_search                 │
│  └── web_fetch                  │
└─────────────────────────────────┘

目录结构

langgraph-runtime/
├── .env                          # 环境变量（API Key 等）
├── requirements.txt              # Python 依赖
├── config/
│   └── agent.json                # LLM 模型配置
├── server/                       # HTTP 服务层
│   ├── main.py                   # FastAPI 入口（含前端托管）
│   └── routes/
│       ├── health.py             # 健康检查
│       └── query.py              # SSE 查询端点（300s 超时）
├── agent/                        # Agent 核心逻辑
│   ├── service.py                # LangGraph 图构建与 astream 流式执行
│   ├── tools.py                  # 工具定义（读文件/跑脚本/搜索/抓取）
│   ├── prompts.py                # System Prompt（引导 Agent 读 SKILL.md）
│   └── skill_router.py           # 问题 → Skill 路由（关键词匹配）
├── static/
│   └── index.html                # 内置聊天前端（暗色主题）
├── skills/                       # Skill 包（每个子目录一个 Skill）
│   ├── deep-research/
│   │   └── SKILL.md
│   ├── research-subagent/
│   │   └── SKILL.md
│   ├── citations/
│   │   └── SKILL.md
│   └── tikhub-api-helper/
│       ├── SKILL.md              # 完整工作流（含 KOL 分析策略）
│       ├── references/
│       │   └── openapi.json      # TikHub 完整 OpenAPI 规范
│       └── scripts/
│           ├── api_client.py     # API 调用客户端（curl + 鉴权）
│           └── api_searcher.py   # API 搜索（中英文混合、别名映射）
└── tests/

快速开始

1. 环境要求

• Python 3.10+
• curl（api_client.py 依赖，用于绕过 Cloudflare TLS 指纹检测）
• 可访问的外部网络（调用 LLM API 和 TikHub API）

2. 安装依赖

cd langgraph-runtime
pip install -r requirements.txt

3. 配置环境变量

编辑项目根目录的 .env 文件：

# LLM API（必填）
# 支持任何 OpenAI 兼容接口
ANTHROPIC_API_KEY=sk-your-api-key-here

# TikHub API Token（tikhub-api-helper Skill 必需）
TIKHUB_TOKEN=your-tikhub-token

# Exa 搜索 API Key（deep-research Skill 的 web_search 必需）
EXA_API_KEY=your-exa-api-key

# 服务端口（可选，默认 3009）
PORT=3009

4. 配置模型参数

编辑 config/agent.json：

{
  "model": "deepseek-chat",
  "base_url": "https://api.deepseek.com",
  "api_key_env": "ANTHROPIC_API_KEY",
  "temperature": 0.3,
  "max_tokens": 4096,
  "recursion_limit": 50
}

已验证可用的模型：DeepSeek (deepseek-chat)，其他支持 tool calling 的 OpenAI 兼容模型也应可用。

5. 启动服务

python -m server.main

看到类似输出即启动成功：

INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:3009

浏览器打开 http://localhost:3009 即可使用内置聊天界面。

前端界面

项目内置了一个单文件暗色主题聊天前端 (static/index.html)，功能包括：

• SSE 流式接收 Agent 回复，实时显示进度
• Markdown 渲染（标题、表格、代码块、列表、引用、粗体/斜体）
• 自动健康检查（每 30 秒检测 /healthz）
• 输入框自动扩展高度，Enter 发送，Shift+Enter 换行
• 响应式布局，支持移动端

前端通过 GET /query/stream?question=... 接口与 Agent 通信，无需额外配置。

API 接口

GET /healthz

健康检查，确认服务状态和 Skill 包是否就绪。

curl http://localhost:3009/healthz

{
  "success": true,
  "status": "ok",
  "skills_dir": "/path/to/skills",
  "skill_count": 4
}

GET /query/stream

SSE 流式查询接口。Agent 收到问题后自动选择 Skill，逐步执行，通过 SSE 事件推送进度和最终结果。

curl -N "http://localhost:3009/query/stream?question=抖音热搜第一是什么"

SSE 事件协议：

监听 final 事件的最简前端代码：

const evtSource = new EventSource("/query/stream?question=" + encodeURIComponent(question));

evtSource.addEventListener("final", (e) => {
  const data = JSON.parse(e.data);
  console.log("回答:", data.answer);
  console.log("耗时:", data.meta.elapsed_ms + "ms");
});

evtSource.addEventListener("done", () => evtSource.close());
evtSource.addEventListener("error", (e) => {
  const data = JSON.parse(e.data);
  console.error("错误:", data.error);
  evtSource.close();
});

Skill 路由

agent/skill_router.py 根据问题中的关键词自动选择 Skill：

路由优先级：社交媒体关键词 > 通用研究。完整关键词列表见 agent/skill_router.py 中的 _SOCIAL_PLATFORMS 集合。

内置 Skill

tikhub-api-helper

通过 TikHub API 获取社交媒体平台数据。

核心工作流：Agent 先读取 SKILL.md 了解工作流 → 用 api_searcher.py 搜索相关 API → 用 api_client.py 调用 API → 多步获取完整数据。

支持平台：抖音、小红书、TikTok、B站、微博、知乎等。

API 搜索工具 (api_searcher.py)：

• 支持中英文混合搜索（如 小红书搜索、抖音热搜、xiaohongshu search）
• 内置平台别名映射（小红书 → xiaohongshu，搜索 → search）
• 多词 AND 匹配，自动展开中文关键词到英文 API 路径
• 支持 tag: 按平台列出所有 API、detail: 查看 API 参数详情

典型查询场景：

目录结构：

skills/tikhub-api-helper/
├── SKILL.md              # 完整工作流（含 KOL 分析策略、API 版本说明）
├── references/
│   └── openapi.json      # TikHub 完整 OpenAPI 规范
└── scripts/
    ├── api_client.py     # API 调用客户端（curl + Bearer 鉴权 + SQLite 日志）
    └── api_searcher.py   # API 搜索（中英文别名、多词匹配、tag 过滤）

deep-research

通用深度研究 Skill，通过网页搜索和抓取回答复杂问题。

触发方式：问题中包含研究类关键词，或未匹配到其他 Skill 时的默认选择

示例问题：

• 今天的东京天气如何
• 对比 React 和 Vue 的优缺点

research-subagent / citations

分别为研究子任务和引用标注的 Skill。当前版本中，deep-research 作为默认入口会整合这些能力。

Agent 工作流程

Agent 遵循 SKILL.md 驱动 的执行模式：

用户问题: "小红书博主阿杰时空扭蛋热门帖子20"
     │
     ▼
1. skill_router 路由
   问题包含"小红书" → 匹配 tikhub-api-helper
     │
     ▼
2. System Prompt 引导
   "先读 SKILL.md，了解工作流"
     │
     ▼
3. Agent 读取 SKILL.md
   read_skill_file("tikhub-api-helper", "SKILL.md")
   了解 KOL 分析工作流和可用 API
     │
     ▼
4. Agent 用 api_searcher.py 搜索 API
   run_skill_script("scripts/api_searcher.py", "小红书搜索笔记")
     │
     ▼
5. Agent 用 api_client.py 调用搜索 API
   run_skill_script("scripts/api_client.py",
     "GET /api/v1/xiaohongshu/app/search_notes keyword=阿杰时空扭蛋 page=1")
   → 从结果中获取 user_id
     │
     ▼
6. Agent 获取用户信息和笔记列表
   并行: get_user_info + get_user_notes
   翻页获取直到满足 20 篇要求
     │
     ▼
7. Agent 按点赞数排序，整理 Top 20 表格
     │
     ▼
8. FastAPI 通过 SSE 推送 final 事件

安全机制

文件读取保护

read_skill_file 使用 Path.resolve() + relative_to() 校验，禁止路径穿越：

# 会被拦截的请求
read_skill_file("tikhub-api-helper", "../../etc/passwd")
read_skill_file("tikhub-api-helper", "../../../.env")

脚本执行限制

run_skill_script 的安全约束：

• 只允许执行 scripts/ 子目录下的 .py 文件
• 使用 shlex.split() 安全解析参数
• 子进程超时 60 秒自动终止
• 工作目录固定在 Skill 根目录
• stderr 中含敏感关键词的行自动脱敏

工具隔离

不同 Skill 可用工具不同：

• tikhub-api-helper：只有 read_skill_file + run_skill_script（不能 web 搜索）
• deep-research：全部 4 个工具（read_skill_file、run_skill_script、web_search、web_fetch）

文件大小限制

• 文本文件读取上限 50,000 字符
• 脚本 stdout 上限 10,000 字符
• web_fetch 内容上限 20,000 字符

添加新 Skill

在 skills/ 下新建目录，至少包含一个 SKILL.md：

skills/my-new-skill/
├── SKILL.md              # 必填，工作流说明
├── references/           # 可选，参考资料
│   └── api-docs.md
└── scripts/              # 可选，可执行脚本
    └── query.py

如果需要自动路由到新 Skill，编辑 agent/skill_router.py 中的关键词映射。如果需要独立的 System Prompt，编辑 agent/prompts.py 中的 build_system_prompt()。

SKILL.md 最佳实践

---
name: my-skill
description: 一句话描述这个 Skill 做什么
---

# My Skill

## 核心工作流
1. 用 api_searcher.py 搜索相关 API
2. 用 api_client.py 调用 API
3. 多步获取完整数据
4. 整理结果返回

## 重要注意事项
（列出 API 版本、必填参数、弃用信息等 Agent 需要知道的关键信息）

## 典型场景工作流
（为每种常见场景提供步骤指引，如 KOL 分析、热搜查询、跨平台对比）

## 关键原则
- 不要在数据不完整时就停止
- 如果 API 返回弃用错误，搜索替代版本

配置项汇总

环境变量 (.env)

模型配置 (config/agent.json)

切换其他 LLM

只需修改 config/agent.json，无需改代码。

// OpenAI
{"model": "gpt-4o", "base_url": "https://api.openai.com/v1", "api_key_env": "OPENAI_API_KEY"}

// Ollama
{"model": "qwen2.5:32b", "base_url": "http://localhost:11434/v1", "api_key_env": "OLLAMA_API_KEY"}

要求：模型必须支持 OpenAI 兼容的 function calling / tool calling 协议。

已知限制

TikHub API 版本变化

• 小红书 Web API (/xiaohongshu/web/*) 全部弃用，必须用 App API (/xiaohongshu/app/*)
• 小红书 search_users 也已弃用，需通过 search_notes 搜索用户名再提取 user_id
• 小红书 App API 的 page 参数为必填
• 其他平台 API 版本可能随 TikHub 更新变化，Agent 会通过 api_searcher.py 动态发现

Agent 执行时间

SSE 超时

• 端点超时设为 300 秒
• 如果 Agent 超时，会在 SSE 流中推送 error 事件

常见问题

Agent 回答"我无法访问实时数据"

检查 TIKHUB_TOKEN 是否有效。TikHub Token 可在 tikhub.io 获取。

小红书搜索返回弃用错误 / Not Found

小红书 Web API 和 search_users 已弃用。SKILL.md 中已记录正确的 App API 路径和替代方案（用 search_notes 搜用户名）。确保 Agent 先读了 SKILL.md。

请求超时 / Recursion limit 错误

• 检查 API 是否返回错误导致 Agent 反复重试
• 在 config/agent.json 中可调整 recursion_limit（默认 50）
• 服务端日志中搜索 [agent-debug] 可查看每步工具调用详情

Agent 数据不完整就停止了

SKILL.md 中有"不要半途而废"的指引。如果仍然出现，可以在 SKILL.md 的对应场景工作流中加强引导。

中文问题乱码

确保 question 参数做了 URL 编码。curl 通常会自动处理，但浏览器 fetch 或 Java 后端需要手动 encodeURIComponent。

特别感谢：

https://linux.do 佬友支持，

https://liang.348349.xyz/ 更多agent项目

把 Agent Skill 包成 API：从本地调用到服务化落地:: https://mp.weixin.qq.com/s/oJOelCN0DgEUmOH1pNEtmA