前面学了输出控制：用 model.withStructuredOutput 来控制输出的结构，它底层会根据模型来决定用 tool 或者 output parser，确保输出一定是符合格式要求的。一般用 withStructuredOutput 就可以了，但当流式返回内容的时候，如果要实现打字机效果，就要直接用 output parser 了，比如 tool 参数的流式打印。
这节我们来练习一下，做两个实战：智能录入和流式版 mini cursor。
实战一：智能录入
场景
当你需要录入信息的时候，之前怎么做呢？一般是在表单里填入信息，点击保存；如果是批量录入，可以上传 excel 来解析录入——这需要你把数据按结构整理好，代码里解析出来保存到数据库。
但在 AI 时代，一般都是智能录入：你只需要给一段文本，让 AI 分析并提取其中的数据，按照结构整理好，然后插入数据库。这是 AI 应用常见功能，就需要用 withStructuredOutput 实现大模型的结构化输出控制。
准备 MySQL
先安装 MySQL 数据库，用 Docker（需要指定 MYSQL_ROOT_PASSWORD 这个环境变量，它是 root 用户的密码）。然后下载一个 GUI 工具连上它，用 MySQL 官方的 MySQL Workbench（https://dev.mysql.com/downloads/workbench/）。
在 output-parser-test 那个项目里写，创建 src/test/create-table.mjs：
import mysql from "mysql2/promise";
async function main() {
const connectionConfig = {
host: "localhost",
port: 3306,
user: "root",
password: "admin",
multipleStatements: true,
};
const connection = await mysql.createConnection(connectionConfig);
try {
// 创建 database
await connection.query(`CREATE DATABASE IF NOT EXISTS hello CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;`);
await connection.query(`USE hello;`);
// 创建好友表
await connection.query(`
CREATE TABLE IF NOT EXISTS friends (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
gender VARCHAR(10), -- 性别
birth_date DATE, -- 出生日期
company VARCHAR(100), -- 公司
title VARCHAR(100), -- 职位
phone VARCHAR(20), -- 当前手机号
wechat VARCHAR(50) -- 微信号
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
`);
// 插入 demo 数据
const insertSql = `
INSERT INTO friends (
name, gender, birth_date, company, title, phone, wechat
) VALUES (?, ?, ?, ?, ?, ?, ?);
`;
const values = [
"王经理", // name
"男", // gender
"1990-01-01", // birth_date
"字节跳动", // company
"产品经理/产品总监", // title
"18612345678", // phone
"wangjingli2024", // wechat
];
const [result] = await connection.execute(insertSql, values);
console.log("成功创建数据库和表，并插入 demo 数据，插入 ID：", result.insertId);
} catch (err) {
console.error("执行出错：", err);
} finally {
await connection.end();
}
}
main().catch((err) => {
console.error("脚本运行失败：", err);
});用 mysql2 的驱动包来连接数据库：
pnpm install mysql2这里创建了 database，创建了一个好友表，然后插入了一条数据。
智能录入
创建 src/test/smart-import.mjs：
import 'dotenv/config';
import { ChatOpenAI } from '@langchain/openai';
import { z } from 'zod';
import mysql from 'mysql2/promise';
// 初始化模型
const model = new ChatOpenAI({
modelName: process.env.MODEL_NAME,
apiKey: process.env.OPENAI_API_KEY,
temperature: 0,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
});
// 定义单个好友信息的 zod schema，匹配 friends 表结构
const friendSchema = z.object({
name: z.string().describe('姓名'),
gender: z.string().describe('性别（男/女）'),
birth_date: z.string().describe('出生日期，格式：YYYY-MM-DD，如果无法确定具体日期，根据年龄估算'),
company: z.string().nullable().describe('公司名称，如果没有则返回 null'),
title: z.string().nullable().describe('职位/头衔，如果没有则返回 null'),
phone: z.string().nullable().describe('手机号，如果没有则返回 null'),
wechat: z.string().nullable().describe('微信号，如果没有则返回 null'),
});
// 定义批量好友信息的 schema（数组）
const friendsArraySchema = z.array(friendSchema).describe('好友信息数组');
// 使用 withStructuredOutput 方法
const structuredModel = model.withStructuredOutput(friendsArraySchema);
// 数据库连接配置
const connectionConfig = {
host: 'localhost',
port: 3306,
user: 'root',
password: 'admin',
multipleStatements: true,
};
async function extractAndInsert(text) {
const connection = await mysql.createConnection(connectionConfig);
try {
// 切换到 hello 数据库
await connection.query(`USE hello;`);
// 使用 AI 提取结构化信息
console.log('🤔 正在从文本中提取信息...\n');
const prompt = `请从以下文本中提取所有好友信息，文本中可能包含一个或多个人信息。请将每个人的信息分别提取出来。
${text}
要求：
1. 如果文本中包含多个人，请为每个人创建一个对象
2. 每个对象包含以下字段：
- 姓名：提取文本中的人名
- 性别：提取性别信息（男/女）
- 出生日期：如果能找到具体日期最好，否则根据年龄描述估算（格式：YYYY-MM-DD）
- 公司：提取公司名称
- 职位：提取职位/头衔信息
- 手机号：提取手机号码
- 微信号：提取微信号
3. 如果某个字段在文本中找不到，请返回 null
4. 返回格式必须是一个数组，即使只有一个人也要放在数组中`;
const results = await structuredModel.invoke(prompt);
console.log(`✅ 提取到 ${results.length} 条结构化信息:`);
console.log(JSON.stringify(results, null, 2));
console.log('');
if (results.length === 0) {
console.log('⚠️ 没有提取到任何信息');
return { count: 0, insertIds: [] };
}
// 批量插入数据库
const insertSql = `
INSERT INTO friends (
name, gender, birth_date, company, title, phone, wechat
) VALUES ?;
`;
const values = results.map((result) => [
result.name,
result.gender,
result.birth_date || null,
result.company,
result.title,
result.phone,
result.wechat,
]);
const [insertResult] = await connection.query(insertSql, [values]);
console.log(`✅ 成功批量插入 ${insertResult.affectedRows} 条数据`);
console.log(` 插入的 ID 范围：${insertResult.insertId} - ${insertResult.insertId + insertResult.affectedRows - 1}`);
return {
count: insertResult.affectedRows,
insertIds: Array.from({ length: insertResult.affectedRows }, (_, i) => insertResult.insertId + i),
};
} catch (err) {
console.error('❌ 执行出错：', err);
throw err;
} finally {
await connection.end();
}
}
// 主函数
async function main() {
// 示例文本（包含多个人的信息）
const sampleText = `我最近认识了几个新朋友。第一个是张总，女的，看起来30出头，在腾讯做技术总监，手机13812345678。第二个是李工，男的，做后端开发，在阿里巴巴，微信号 likegong2024。还有一个是王老师，女博士，在北大当老师。`;
console.log('📝 输入文本:');
console.log(sampleText);
console.log('');
try {
const result = await extractAndInsert(sampleText);
console.log(`\n🎉 处理完成！成功插入 ${result.count} 条记录`);
console.log(` 插入的 ID：${result.insertIds.join(', ')}`);
} catch (error) {
console.error('❌ 处理失败：', error.message);
process.exit(1);
}
}
main();给一段无规则文本，用大模型提取结构化的信息：结构用 withStructuredOutput 指定，要求提取一个数组，数组里是好友对象的信息。然后把数组里的结构化数据批量插入数据库表。
跑一下：大模型把张总、李工、王老师的信息都提取出来了，并且批量插入了数据库。这样，我们就实现了智能录入的功能。
补充：JSON Schema 机制
前面讲 withStructuredOutput 底层是 tool、output parser，其实还有第三种特性：JSON Schema。创建 src/structured-json-schema.mjs：
import 'dotenv/config';
import { ChatOpenAI } from '@langchain/openai';
import { z } from 'zod';
import { zodToJsonSchema } from "zod-to-json-schema";
import { HumanMessage, SystemMessage } from '@langchain/core/messages';
const scientistSchema = z.object({
name: z.string().describe("科学家的全名"),
birth_year: z.number().describe("出生年份"),
field: z.string().describe("主要研究领域"),
achievements: z.array(z.string()).describe("主要成就列表")
}).strict();
// 将 Zod 转换为原生的 JSON Schema 格式
const nativeJsonSchema = zodToJsonSchema(scientistSchema);
const model = new ChatOpenAI({
modelName: "qwen-max",
temperature: 0,
apiKey: process.env.OPENAI_API_KEY,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
modelKwargs: { // 通过 modelKwargs 传入原生参数
response_format: {
type: "json_schema",
json_schema: {
name: "scientist_info",
strict: true,
schema: nativeJsonSchema // 这里的 nativeJsonSchema 就是转换后的对象
}
}
}
});
async function testNativeJsonSchema() {
console.log("🧪 测试原生 JSON Schema 模式...\n");
const res = await model.invoke([
new SystemMessage("你是一个信息提取助手，请直接返回 JSON 数据。"),
new HumanMessage("介绍一下杨振宁")
]);
console.log("\n✅ 收到响应（纯净 JSON）:");
console.log(res.content);
const data = JSON.parse(res.content);
console.log("\n📋 解析后的对象:");
console.log(data);
}
testNativeJsonSchema().catch(console.error);指定大模型的输出格式为 json_schema，它就会按照这个格式输出。安装依赖：
pnpm install zod-to-json-schemaJSON Schema 就和 tool 的 args 一样，都是大模型层面支持的，会保证按照这个格式来返回，如果格式不对，会在模型层面重新生成正确的返回。
也就是说，withStructuredOutput 底层是 tool、json schema、output parser 这三者。当然，平时开发用 withStructuredOutput 就可以了，这个 API 会根据模型自动选择对应的实现。
实战二：流式版 mini cursor
还记得之前做的 mini-cursor 么？当时等了好久，大概一分钟才显示写入成功。其实这时候一直在生成代码内容，只不过我们没做流式打印，只能干等。我们学完流式 + output parser 之后，就可以优化了。
先想想之前不用流式是什么流程：传入 SystemMessage 和 HumanMessage，调用大模型之后返回 AIMessage；把这个 AIMessage 加入 memory；之后根据 AIMessage 中的 tool_calls 信息调用 tool，执行结果封装成 ToolMessage 放入 memory；直到不再返回带 tool_calls 信息的 AIMessage，就代表循环结束。
那如果改成流式返回的话，难点在哪呢？
- 难点一：返回的 AIMessage 是 chunk。我们要把
AIMessageChunk拼接成完整的 AIMessage 才能放入 Memory 再次调用大模型——用它的concat方法即可。流式返回一个个 AIMessageChunk，调用concat方法合并一下，流式结束就拿到了完整的 AIMessage。 - 难点二：现在返回的是 tool_call_chunks 怎么办？前面我们用
JsonOutputToolsParser把它转成了 tool_calls。原始的 stream 里的 tool_call_chunks 是片段信息，并不是一个合法的 json；而用了JsonOutputToolsParser之后，拿到的就是积累的片段拼起来的 json 了。所以我们不用自己解析这些片段，直接用JsonOutputToolsParser解析之后取某个属性就行。
按照这个思路来实现新版 mini cursor，创建 src/test/mini-cursor.mjs：
import 'dotenv/config';
import { ChatOpenAI } from '@langchain/openai';
import { HumanMessage, SystemMessage, ToolMessage } from '@langchain/core/messages';
import { InMemoryChatMessageHistory } from '@langchain/core/chat_history';
import { JsonOutputToolsParser } from '@langchain/core/output_parsers/openai_tools';
import { executeCommandTool, listDirectoryTool, readFileTool, writeFileTool } from './all-tools.mjs';
import chalk from 'chalk';
const model = new ChatOpenAI({
modelName: "qwen-plus",
apiKey: process.env.OPENAI_API_KEY,
temperature: 0,
configuration: {
baseURL: process.env.OPENAI_BASE_URL,
},
});
const tools = [
readFileTool,
writeFileTool,
executeCommandTool,
listDirectoryTool,
];
// 绑定工具到模型
const modelWithTools = model.bindTools(tools);
// Agent 执行函数
async function runAgentWithTools(query, maxIterations = 30) {
const history = new InMemoryChatMessageHistory();
await history.addMessage(new SystemMessage(`你是一个项目管理助手，使用工具完成任务。
当前工作目录: ${process.cwd()}
工具：
1. read_file: 读取文件
2. write_file: 写入文件
3. execute_command: 执行命令（支持 workingDirectory 参数）
4. list_directory: 列出目录
重要规则 - execute_command：
- workingDirectory 参数会自动切换到指定目录
- 当使用 workingDirectory 时，绝对不要在 command 中使用 cd
- 错误示例: { command: "cd react-todo-app && pnpm install", workingDirectory: "react-todo-app" }
- 正确示例: { command: "pnpm install", workingDirectory: "react-todo-app" }
重要规则 - write_file：
- 当写入 React 组件文件（如 App.tsx）时，如果存在对应的 CSS 文件（如 App.css），在其他 import 语句后加上 import 该 CSS 文件
`));
await history.addMessage(new HumanMessage(query));
for (let i = 0; i < maxIterations; i++) {
console.log(chalk.bgGreen(`⏳ 正在等待 AI 思考...`));
// 获取当前消息历史
const messages = await history.getMessages();
const rawStream = await modelWithTools.stream(messages);
// 准备一个空的容器来拼接完整的 AIMessage
let fullAIMessage = null;
// 准备一个 tool_call_chunks 的 JSON 增量解析器
const toolParser = new JsonOutputToolsParser();
// 记录每个工具调用已打印的长度（用 id 或 filePath 作为 key）
const printedLengths = new Map();
console.log(chalk.bgBlue(`\n🚀 Agent 开始思考并生成流...\n`));
for await (const chunk of rawStream) {
// 这里的 chunk 是 AIMessageChunk，把它拼接起来
fullAIMessage = fullAIMessage ? fullAIMessage.concat(chunk) : chunk;
let parsedTools = null;
try {
parsedTools = await toolParser.parseResult([{ message: fullAIMessage }]);
} catch (e) {
// 解析失败说明 JSON 还不完整，忽略错误继续累积
}
if (parsedTools && parsedTools.length > 0) {
for (const toolCall of parsedTools) {
if (toolCall.type === 'write_file' && toolCall.args?.content) {
const toolCallId = toolCall.id || toolCall.args.filePath || 'default';
const currentContent = String(toolCall.args.content);
const previousLength = printedLengths.get(toolCallId);
if (previousLength === undefined) {
printedLengths.set(toolCallId, 0);
console.log(
chalk.bgBlue(`\n[工具调用] write_file("${toolCall.args.filePath}") - 正在生成内容...`)
);
}
if (currentContent.length > previousLength) {
const newContent = currentContent.slice(previousLength);
process.stdout.write(newContent);
printedLengths.set(toolCallId, currentContent.length);
}
}
}
} else {
// 当前还没有解析出工具调用时，如果有文本内容就直接输出
if (chunk.content) {
process.stdout.write(
typeof chunk.content === 'string'
? chunk.content
: JSON.stringify(chunk.content),
);
}
}
}
// 此时 fullAIMessage 已经完美还原，直接存入 history
await history.addMessage(fullAIMessage);
console.log(chalk.green('\n✅ 消息已完整存入历史'));
// 检查是否有工具调用
if (!fullAIMessage.tool_calls || fullAIMessage.tool_calls.length === 0) {
console.log(`\n✨ AI 最终回复:\n${fullAIMessage.content}\n`);
return fullAIMessage.content;
}
// 执行工具调用
for (const toolCall of fullAIMessage.tool_calls) {
const foundTool = tools.find((t) => t.name === toolCall.name);
if (foundTool) {
const toolResult = await foundTool.invoke(toolCall.args);
await history.addMessage(
new ToolMessage({
content: toolResult,
tool_call_id: toolCall.id,
}),
);
}
}
}
const finalMessages = await history.getMessages();
return finalMessages[finalMessages.length - 1].content;
}
const case1 = `创建一个功能丰富的 React TodoList 应用：
1. 创建项目： echo -e "n\nn" | pnpm create vite react-todo-app --template react-ts
2. 修改 src/App.tsx，实现完整功能的 TodoList：
- 添加、删除、编辑、标记完成
- 分类筛选（全部 / 进行中 / 已完成）
- 统计信息显示
- localStorage 数据持久化
3. 添加复杂样式：
- 渐变背景（蓝到紫）
- 卡片阴影、圆角
- 悬停效果
4. 添加动画：
- 添加 / 删除时的过渡动画
- 使用 CSS transitions
5. 列出目录确认
注意：使用 pnpm，功能要完整，样式要美观，要有动画效果
去掉 main.tsx 里的 index.css 导入
之后在 react-todo-app 项目中：
1. 使用 pnpm install 安装依赖
2. 使用 pnpm run dev 启动服务器
`;
try {
await runAgentWithTools(case1);
} catch (error) {
console.error(`\n❌ 错误: ${error.message}\n`);
}其实你理解了刚才的两个点——拼接 AIMessageChunk 成完整 AIMessage、用 JsonOutputToolsParser 来解析 tool_call_chunks——就很容易理解上面的代码。
- 我们用
InMemoryChatMessageHistory来管理 memory，不是直接用 messages 数组了 - 拼接：
fullAIMessage = fullAIMessage.concat(chunk)，当 chunk 遍历完，自然也就拼好了，这时候就放入 memory - 解析：调用
toolParser.parseResult([{ message: fullAIMessage }])，传入我们 concat 好的部分 AIMessage - 增量打印：加了一个
Map来记录已经打印过的长度，下次slice一下继续打印后面的。第一次打印工具调用的日志，在 Map 里记录长度为 0，后续每次打印截取掉这个长度之后的 args 信息，这样就能实现增量打印的效果
用到的 all-tools.mjs 从之前 tool-test 里复制，再安装下依赖：
pnpm install chalk跑一下：现在的 mini cursor 就能流式打印生成的代码了。流式打印生成的代码，等完整之后再一次写入即可——写入的内容从拼接完整的 AIMessage 取 tool_calls 的参数信息；前面用 JsonOutputToolsParser 解析 tool_call_chunks 拼起来的 tool_calls 的 args 只是流式打印用，调用工具的时候还是直接从 AIMessage 取完整的。
虽然只是加了一个流式，但是代码改动还挺多的：之前是完整的 AIMessage 和 tool_calls，现在需要自己对 AIMessageChunk 做 concat，以及用 JsonOutputToolsParser 解析 tool_call_chunks。
常见问题
MySQL 创建表报 SQL 语法错误？ 检查 multipleStatements: true 是否设置，以及 SQL 语句的完整性。如果 Docker 搜索 MySQL 镜像结果为 0，可以命令行 docker pull mysql 或科学上网。
大模型把"28 岁"推成"1996-01-01"？ 大模型训练有截止时间，它不知道当前时间。可以写动态提示词（"当前时间是 ${Date.now()}"），或者写一个 tool 来计算生日。
有时第一轮或第二轮就结束了？ 加一个 content 跟工具调用都为空时的判断，强制再触发一轮。
国产模型（如 deepseek）不支持 json_schema？ 需要：1. 关闭思考模式；2. 用 z.object({ friends: z.array(friendSchema) }) 包裹；3. 设置 method: "functionCalling"。如果用 jsonObject 模式，需要在 prompt 里告诉 LLM key 的英文（否则会输出中文 key）。
zodToJsonSchema 报错或用不了？ 新版本可以直接用 scientistSchema.toJSONSchema()，不需要额外装 zod-to-json-schema 包。
为什么不直接打印 tool_call_chunks 要绕一圈 JsonOutputToolsParser？ 因为 tool_call_chunks 是不完整的 json 片段，你不知道该打印哪部分；比如参数的 json 有 4 个属性，你要打印的是第三个属性，直接打印 chunk 做不到，正则也不靠谱。parser 才能拿到完整结构。
流式过程中工具参数能直接用吗？ 流式过程中 parser 解析出的 args 可能还不完整（同一个 tool 在 parseResult 时会经历成功→失败→成功多次循环），调用工具时还是从完整拼接后的 AIMessage 取 tool_calls 参数。
总结
这节我们做了大模型输出控制的两个小实战：
- 智能录入：常见需求，调用大模型对一段文本做解析，返回结构化的数据，一般用
model.withStructuredOutput，之后存入数据库即可 - 流式版 mini cursor：主要要流式打印 tool 的参数，需要做好 AIMessageChunk 的 concat，以及用
JsonOutputToolsParser做 tool_call_chunks 的解析，之后增量打印
此外，我们还补充学习了 withStructuredOutput 底层的另一个 JSON Schema 机制。当然，平时做结构化直接用 withStructuredOutput 就行，底层会自动根据模型来选择 tool、json schema 或者 output parser。
常见的输出控制需求就这两种：结构化输出、流式输出 + tool 参数解析。