用 TypeScript 重写 mini-agent

:::info 学习目标完成本篇后你能够：用 TypeScript 独立实现一个带工具循环的 agent；对比 Python 版与 TS 版的表达差异；为它加上 SSE 流式输出。前置：第 1~9 篇完成；读过 Agent 专栏第 4 篇更佳。预计时长：90 分钟。 :::

Agent 专栏第 4 篇用 Python 手写了一个 100 行的 agent。今天用 TypeScript 重写——逻辑完全一致，顺便把前九篇的零件全部用上：类型（1）、模块（2）、async（3）、事件（4）、schema 校验（6）、SSE（8）。

完整实现

// mini_agent.ts —— Node 22+，npx tsx mini_agent.ts "任务" 直接运行
import { spawnSync } from 'node:child_process'

const API = 'https://api.deepseek.com/chat/completions'
const KEY = process.env.DEEPSEEK_API_KEY ?? ''

interface ChatMessage {
  role: 'system' | 'user' | 'assistant' | 'tool'
  content: string | null
  tool_calls?: { id: string; function: { name: string; arguments: string } }[]
  tool_call_id?: string
}

const TOOLS = [{
  type: 'function', function: {
    name: 'run_bash',
    description: '在工作目录执行一条 bash 命令，返回 stdout+stderr。用于查看文件、搜索、跑测试。',
    parameters: { type: 'object', properties: {
      command: { type: 'string', description: '要执行的命令' },
    }, required: ['command'] },
  },
}] as const

const SYSTEM = '你是一个谨慎的编码助手。先探索再行动；任务完成后直接总结，不要调用工具。'

async function chat(messages: ChatMessage[]): Promise<ChatMessage> {
  const res = await fetch(API, {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${KEY}` },
    body: JSON.stringify({ model: 'deepseek-chat', messages, tools: TOOLS, temperature: 0.2 }),
  })
  if (!res.ok) throw new Error(`API ${res.status}: ${await res.text()}`)
  const data = await res.json()
  return data.choices[0].message as ChatMessage
}

function execute(name: string, args: { command?: string }): string {
  if (name === 'run_bash' && args.command) {
    const p = spawnSync(args.command, { shell: true, encoding: 'utf8', timeout: 60_000 })
    return ((p.stdout ?? '') + (p.stderr ?? '') || '(no output)').slice(0, 4000)
  }
  return `unknown tool: ${name}`
}

async function run(task: string, maxSteps = 25): Promise<string> {
  const messages: ChatMessage[] = [
    { role: 'system', content: SYSTEM },
    { role: 'user', content: task },
  ]
  for (let step = 1; step <= maxSteps; step++) {
    const reply = await chat(messages)
    messages.push(reply)
    const calls = reply.tool_calls ?? []
    if (calls.length === 0) return reply.content ?? '(空回复)'
    console.log(`[step ${step}] ${reply.content ?? '(thinking…)'}`)
    for (const call of calls) {
      let args: Record<string, unknown> = {}
      try { args = JSON.parse(call.function.arguments) }          // wire 格式：字符串要自己 parse
      catch { args = {} }
      console.log(`  → ${call.function.name}(${JSON.stringify(args)})`)
      let result: string
      try { result = execute(call.function.name, args) }
      catch (e) { result = `Error: ${(e as Error).message}` }     // 失败也回填
      messages.push({ role: 'tool', tool_call_id: call.id, content: result })
    }
  }
  return '达到步数上限，任务未确认完成'
}

run(process.argv.slice(2).join(' ') || '统计当前目录文件数')
  .then(console.log)

注：上面 execute 里演示性的 Bun 分支可以删掉；正式版本用 node:child_process 的 spawnSync 即可（保留 timeout 与输出截断）。

TS 版的循环全景

Python 版 vs TypeScript 版：逐点对照

维度	Python 版（第 4 篇）	TypeScript 版（本篇）	一致的内核
HTTP	`urllib.request`	全局 `fetch`	一样的 JSON 请求体
工具 schema	dict 字面量	`as const` 对象字面量	同一份 JSON 结构
参数解析	`json.loads`	`JSON.parse`	模型产出 JSON 字符串，自己解析
错误回填	try/except → “Error: …”	try/catch → “Error: …”	失败回流，模型自纠
截断	`[:4000]` 切片	`.slice(0, 4000)`	防单结果吃光窗口
类型安全	无（运行时校验）	ChatMessage 接口编译期保障	TS 版多了编译期防线

结论：agent 循环是语言无关的。第 4 篇那张流程图，换个语言重画一遍而已。

加餐：接入第 8 篇的 SSE

把模拟上游（第 8 篇的 /v1/chat）换成真实 API 的 stream: true，用 for await 消费 SSE：

const stream = await fetch(API, { ...opts, body: JSON.stringify({ ...body, stream: true }) })
for await (const chunk of stream.body!) {
  for (const line of chunk.toString().split('\n')) {
    if (line.startsWith('data: ') && line !== 'data: [DONE]') {
      const delta = JSON.parse(line.slice(6))
      process.stdout.write(delta.choices?.[0]?.delta?.content ?? '')
    }
  }
}

常见踩坑

fetch 不存在——Node 18 以下没有全局 fetch，升级 Node 或引入 undici；
tool_calls 可能为 undefined——非工具轮次没有这个字段，访问前判空（TS 类型已经提醒你了）；
环境变量忘配置——DEEPSEEK_API_KEY 为空时请求 401，启动时先检查并给出友好提示。

随堂练习（带验收标准）

跑通三个任务（读/执行/混合）。验收：与 Python 版行为一致、步数相当；
加 SSE 流式输出。验收：终端逐字打印回复；
进阶：把工具执行改成第 8 篇的异步版本（spawn + promise），支持 5 秒超时。