获取工具调用信息

更新工具调用信息（累积式） Args: tool_id: 工具调用 ID name: 工具名称 args: 工具参数 args_complete: 参数是否完整。区分"参数为空字典"和"

格式化为 Claude Code 风格的紧凑格式：ToolName(arg1, arg2, ...) Args: name: 工具名称 args: 工具参数字典 Returns: 格式化的字符串，如 "Bash(git s

处理单个流式事件，更新内部状态 Args: event: 流式事件字典 Returns: 事件类型

解析文件路径，处理相对路径和 ~ 展开 Args: file_path: 文件路径（绝对或相对，支持 ~ 表示用户主目录） working_directory: 工作目录 Returns: 解析后的绝对路径

检测内容类型

累积 input_json_delta 片段 LangChain 流式传输中，args 可能以 input_json_delta 形式分批到达。 index 用于处理并行工具调用的情况。

判断工具输出是否表示成功执行 基于 [OK]/[FAILED] 前缀判断。 Args: content: 工具输出内容 Returns: True 如果成功执行

事件级流式输出，支持 thinking 和 token 级流式 Args: message: 用户消息 thread_id: 会话 ID Yields: 事件字典，格式如下:

Create FastAPI app with injectable agent provider (for tests).

检查工具调用是否准备好发送（有 name 且未发送即可）

直接执行 bash 命令的测试辅助函数（复制 tools.py 中的逻辑）

Level 1: 扫描所有 Skills 元数据 遍历 skill_paths，查找包含 SKILL.md 的目录， 解析 YAML frontmatter 提取 name 和 description。 Returns:

Parse comma-separated origins from env.

标记已发送

判断内容是否表示成功执行

截断内容

最终化所有工具调用：解析累积的 JSON 片段并标记参数完整

智能格式化工具结果 Args: name: 工具名称 content: 工具输出内容 max_length: 最大显示长度 compact: 是否使用紧凑的树形格式（Claude Code 风格） Retu

工具调用事件

解析 provider:model 形式的模型字符串

统计内容行数

创建流式显示的布局 Args: thinking_text: 当前累积的 thinking 文本 response_text: 当前累积的响应文本 tool_calls: 工具调用列表 tool_results: 工

将输出格式化为树形结构（Claude Code 风格） Args: lines: 输出行列表 max_lines: 最大显示行数 indent: 缩进字符 Returns: 格式化的树形输出字符串

获取发现的 Skills 列表 用于演示 Level 1 的 Skills 发现过程。

解析当前模型配置

Encode one SSE frame.

构建包含 Skills 列表的 system prompt 这是 Level 1 的核心输出：将所有 Skills 的元数据 注入到 system prompt 中。 Args: base_prompt:

检查是否配置了当前 provider 的 API 认证

创建流式显示的布局

获取所有工具调用（包括已发送的）

获取用于 create_streaming_display 的参数

获取所有未发送的工具调用

检查 args 是否有内容 修复空字典 falsy 问题：空字典 {} 在 Python 中是 falsy， 但对于工具调用来说，空字典表示无参数，是合法的。 Args: args: 工具参数，可能是 None、{} 或包含参数的字典

Level 2: 加载 Skill 完整内容 读取 SKILL.md 的完整指令，以及其他 .md 文件和脚本列表。 这是 load_skill tool 的核心实现。 Args: skill_name:

文本内容事件

思考内容事件

截断内容到指定长度 Args: content: 要截断的内容 max_length: 最大长度 suffix: 截断后添加的后缀 Returns: 截断后的内容

按行数截断内容，并返回剩余行数 Args: content: 要截断的内容 max_lines: 最大显示行数 Returns: (截断后的内容, 剩余行数)

提取状态标记后的内容体

获取 provider 对应的 API 认证信息 支持多种认证方式： 1. 通用配置：MODEL_API_KEY / MODEL_BASE_URL 2. Provider 专属配置 3. OpenAI 兼容代理场景下，允许复用 ANTHROPIC_AUTH

根据模型名推断 provider

检查是否是 JSON

规范化 OpenAI SDK base_url，避免传入完整 endpoint 后被重复拼接。

标准化 provider 名称

解析当前请求应使用的 provider

交互式对话模式，支持流式输出和 thinking 显示 Args: enable_thinking: 是否启用 thinking 显示

列出发现的 Skills

显示 system prompt（演示 Level 1）

显示最终结果（非流式） Args: state: 流式处理状态 thinking_max_length: thinking 最大显示长度 tool_result_max_length: 工具结果最大显示长度 args

完成事件

错误事件

格式化工具结果

格式化工具参数显示 Args: args: 工具参数字典 max_length: 最大显示长度 Returns: Rich 可渲染元素列表

从结果中提取最后的 AI 响应文本 Args: result: invoke 或 stream 的结果 Returns: AI 响应文本

同步调用 Agent Args: message: 用户消息 thread_id: 会话 ID（用于多轮对话） Returns: Agent 响应

流式调用 Agent (state 级别) Args: message: 用户消息 thread_id: 会话 ID Yields: 流式响应块 (完整状态更新)

生成 system prompt 中的单行描述

工具结果事件

构建 system prompt 这是 Level 1 的核心：将所有 Skills 的元数据注入到 system prompt。 每个 skill 约 100 tokens，启动时一次性加载。

创建 LangChain Agent 使用 LangChain 1.0 的 create_agent API: - model: 可以是字符串 ID 或 model 实例 - tools: 工具列表 - system

从 OpenAI chunk 的 usage_metadata 中提取 reasoning token 数量

格式化普通文本输出

检查是否是错误内容

检查是否是 Markdown

解析布尔环境变量

解析 SKILL.md 的 YAML frontmatter SKILL.md 格式： --- name: skill-name description: Brief description

处理 chunk 的 content

处理 chunk.tool_calls - 立即发送 tool_call 事件 避免重复发送（同一 tool 可能通过 tool_use block 已发送）。

处理工具结果

解析模型名称，兼容旧环境变量

交互式对话（流式输出版本）

清空追踪器

执行单次请求，支持流式输出和 thinking 显示 Args: prompt: 用户请求 enable_thinking: 是否启用 thinking 显示

演示 Level 1: Skills 发现 启动时自动扫描 Skills 目录，解析 YAML frontmatter， 将元数据（name + description）注入 system prompt。 Token 消耗：约 100 tokens/skill

演示 Level 2: 加载 Skill 详细指令 当用户请求匹配 Skill 描述时， LLM 调用 load_skill 工具读取 SKILL.md 完整内容。 Token 消耗：约 5k tokens

演示 Level 3: 脚本执行 LLM 按照 Skill 指令调用 bash 工具执行脚本。 脚本代码不进入上下文，只有输出进入。 这是 Skills 高效的关键：大量代码被封装在脚本中， 不占用上下文空间。

演示流式输出和 thinking 功能

Token 消耗分析

便捷函数：发现所有 Skills

发送所有待处理的工具调用并标记

使用 news-extractor Skill 提取文章