Dedalus Docs / Home

LLM 会生成文本，而应用程序需要数据结构。结构化输出弥合了这一差距——你定义一个模式（在 Python 中使用 Pydantic，在 TypeScript 中使用 Zod 或 Effect Schema），Dedalus SDK 会确保响应完全符合该模式，并提供完整的类型安全。这对于构建可靠的应用至关重要。你不再需要解析非结构化文本再寄希望于结果正确，而是可以获得经过验证、可被代码信任的对象。

提取类型化数据

定义一个 schema，调用 .parse()，即可获取通过校验的对象。

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": "Return 3 upcoming basketball events near San Francisco as JSON.",
        }],
        response_format=EventsResponse,
    )

    parsed: EventsResponse = completion.choices[0].message.parsed
    print(parsed)

if __name__ == "__main__":
    asyncio.run(main())

高级用法

本节是一个供你快速浏览、随时查阅的参考，按循序渐进的顺序组织：

Client .parse()（非流式传输，类型化输出）
Client .stream()（流式传输，类型化输出）
Runner response_format（智能体 / tool 循环中的类型化输出）
Schema 与模式（可选字段、嵌套模型、枚举 / 联合）
结构化的 tool 调用（当你需要确定性的工具调用时）

Client API（参考）

Client 提供了三种用于结构化输出的方法：

.parse() - 非流式传输，使用类型安全的 schema
.stream() - 流式传输，使用类型安全的 schema（作为上下文管理器）
.create() - 仅支持基于字典的 schema

TypeScript 设置

TypeScript schema 辅助工具是可选的对等依赖。安装你想要使用的校验库：

bun install zod
# 或
bun install effect

`.parse()`（非流式传输）

这与上面的渐进式示例遵循相同的模式，只是改用更偏向“应用程序编程接口 API 参考”风格的写法再展示一遍。

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[
            {
                "role": "user",
                "content": (
                    "Return 3 upcoming basketball events near San Francisco as JSON. "
                    "Use ISO dates (YYYY-MM-DD)."
                ),
            }
        ],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],  # 通过 Ticketmaster 发现赛事
    )

    parsed = completion.choices[0].message.parsed
    print(parsed)

if __name__ == "__main__":
    asyncio.run(main())

TypeScript：使用 Effect Schema 的相同示例

import Dedalus from 'dedalus-labs';
import { effectResponseFormat } from 'dedalus-labs/helpers/effect';
import * as Schema from 'effect/Schema';

const client = new Dedalus();

const Event = Schema.Struct({
  name: Schema.String,
  city: Schema.String,
  date: Schema.String,
});

async function main() {
  const completion = await client.chat.completions.parse({
    model: 'openai/gpt-5.2',
    messages: [
      {
        role: 'user',
        content:
          '以 JSON 格式返回 3 场旧金山附近即将举行的篮球赛事。使用 ISO 日期 (YYYY-MM-DD)。',
      },
    ],
    response_format: effectResponseFormat(
      Schema.Struct({ query: Schema.String, events: Schema.Array(Event) }),
      'events_response',
    ),
    mcpServers: ['windsor/ticketmaster-mcp'],
  });

  console.log(completion.choices[0]?.message.parsed);
}

main();

`.stream()`（流式传输）

当你既需要流式交互体验又需要带类型的最终结果时，使用这个方法。

不同语言的流式传输辅助方式不同：

Python：将 .stream(...) 作为上下文管理器使用，并读取带类型的流式事件。
TypeScript：通过 create({ stream: true, ... }) 以流式方式传输 token，然后用 Zod/Effect 校验最终的 JSON。

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    # 使用上下文管理器进行流式传输
    async with client.chat.completions.stream(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "Return 3 upcoming basketball events near San Francisco as JSON. "
                "Use ISO dates (YYYY-MM-DD)."
            ),
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    ) as stream:
        # 事件到达时实时处理
        async for event in stream:
            if event.type == "content.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "content.done":
                # 在 content.done 时可获得快照（带类型）
                print(f"\nParsed events: {len(event.parsed.events)}")

        # 获取最终解析结果
        final = await stream.get_final_completion()
        parsed = final.choices[0].message.parsed
        print(f"\nFinal events: {len(parsed.events)}")

if __name__ == "__main__":
    asyncio.run(main())

可选字段

在 Python 中使用 Optional[T]，在 Zod 中使用 .nullable()，在 Effect 中使用 Schema.NullOr(...) 来表示可为空的字段：

在 OpenAI 的严格模式下，每个字段都必须是必填项。将语义上的“可选”值建模为可空（nullable）字段。

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str
    price_usd: int | None = None  # 当价格未知时为 null

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "Return 3 upcoming basketball events near San Francisco as JSON. "
                "Include price_usd if known; otherwise null. Use ISO dates (YYYY-MM-DD)."
            ),
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, e.price_usd)

if __name__ == "__main__":
    asyncio.run(main())

TypeScript (Effect)：可空字段

import * as Schema from 'effect/Schema';

const Event = Schema.Struct({
  name: Schema.String,
  city: Schema.String,
  date: Schema.String,
  price_usd: Schema.NullOr(Schema.Number),
});

const EventsResponse = Schema.Struct({
  query: Schema.String,
  events: Schema.Array(Event),
});

对于结构化输出，请避免使用 Schema.optional(...)——改用 Schema.NullOr(...)。

Schema 与模式

嵌套模型

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Venue(BaseModel):
    name: str
    address: str | None = None
    city: str

class Event(BaseModel):
    name: str
    date: str
    venue: Venue

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "以 JSON 格式返回 3 个旧金山附近即将发生的篮球赛事。"
                "每个赛事必须包含一个嵌套的 venue 对象，其中包含 name、city 和 address 字段（未知时为 null）。"
                "日期使用 ISO 格式 (YYYY-MM-DD)。"
            )
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, "→", e.venue.name)

结构化工具调用（高级）

定义具备自动参数解析的类型安全工具：

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class SearchEventsArgs(BaseModel):
    city: str
    month: str
    max_results: int = 5

async def main():
    client = AsyncDedalus()

    tools = [
        {
            "type": "function",
            "function": {
                "name": "search_events",
                "description": "Search for events in a city during a month.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string"},
                        "month": {"type": "string", "description": "YYYY-MM"},
                        "max_results": {"type": "integer", "default": 5},
                    },
                    "required": ["city", "month"],
                    "additionalProperties": False,
                },
                "strict": True,
            }
        }
    ]

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": "Call search_events for San Francisco in 2026-01.",
        }],
        tools=tools,
        tool_choice={"type": "tool", "name": "search_events"},
    )

    message = completion.choices[0].message
    if message.tool_calls:
        tool_call = message.tool_calls[0]
        print(f"Tool called: {tool_call.function.name}")
        print(f"Parsed args: {tool_call.function.parsed_arguments}")

if __name__ == "__main__":
    asyncio.run(main())

如果你需要确定性的工具调用，将 tool_choice 设置为以下对象形式之一： { type: 'auto' }（由模型决定）、{ type: 'any' }（强制要求进行一次工具调用）、{ type: 'tool', name: 'search_events' }（强制调用特定工具）、{ type: 'none' }（禁用工具）。传入 OpenAI 字符串形式（例如 tool_choice: 'required'）会在模式验证时失败，并返回 422。

TypeScript：使用 Effect Schema 定义 tool 参数

import { effectFunction } from 'dedalus-labs/helpers/effect';
import * as Schema from 'effect/Schema';

const SearchEventsTool = effectFunction({
  name: 'search_events',
  parameters: Schema.Struct({
    city: Schema.String,
    month: Schema.String, // YYYY-MM
    max_results: Schema.NullOr(Schema.Number),
  }),
  description: 'Search for events in a city during a month.',
});

tool 参数必须是一个对象 schema（使用 Schema.Struct({ ... })）。

枚举与联合类型

import asyncio
from typing import Literal
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str
    category: Literal["sports", "music", "theater", "other"]
    ticket_status: Literal["available", "sold_out", "unknown"]

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-5.2",
        messages=[{
            "role": "user",
            "content": (
                "Return 3 upcoming events near San Francisco as JSON. "
                "Each event must include category (sports/music/theater/other) and ticket_status (available/sold_out/unknown). "
                "Use ISO dates (YYYY-MM-DD)."
            )
        }],
        response_format=EventsResponse,
        mcp_servers=["windsor/ticketmaster-mcp"],
    )

    parsed = completion.choices[0].message.parsed
    for e in parsed.events:
        print(e.name, e.category, e.ticket_status)

if __name__ == "__main__":
    asyncio.run(main())

DedalusRunner 应用程序编程接口（API）

Runner 支持 response_format，并可自动进行 schema 转换：

import asyncio

from dedalus_labs import AsyncDedalus, DedalusRunner
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

def as_bullets(items: list[str]) -> str:
    """Format items as a bulleted list."""
    return "\n".join(f"• {item}" for item in items)

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input=(
            "Find me the nearest basketball games in January in San Francisco using Ticketmaster. "
            "Then call as_bullets with a list of items (one per event: name, city, date)."
        ),
        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/ticketmaster-mcp"],  # 通过 Ticketmaster 发现赛事
        tools=[as_bullets],
        response_format=EventsResponse,
        max_steps=5,
    )

    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

.create() vs .parse() vs .stream()

Method	Schema 支持	流式传输	使用场景
`.create()`	仅 dict	✓	手动维护 JSON Schema
`.parse()`	Pydantic/Zod/Effect	❌	类型安全的非流式传输
`.stream()`	Pydantic/Zod/Effect	✓	类型安全的流式传输

.create() 期望接收一个普通的 JSON Schema 对象。不要直接传入 Pydantic 模型、Zod schema 或 Effect schema。

“流式传输 + 类型化输出”取决于语言：

Python：.stream(...) 会依次产生带类型的事件，以及一个带类型的最终快照。
TypeScript：以流式传输方式输出 tokens，并使用 Zod/Effect 校验最终 JSON。

错误处理

import asyncio
from typing import Any
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Event(BaseModel):
    name: str
    city: str
    date: str

class EventsResponse(BaseModel):
    query: str
    events: list[Event]

async def main():
    client = AsyncDedalus()

    try:
        completion = await client.chat.completions.parse(
            model="openai/gpt-5.2",
            messages=[{
                "role": "user",
                "content": (
                    "Return 3 upcoming basketball events near San Francisco as JSON. "
                    "Use ISO dates (YYYY-MM-DD)."
                ),
            }],
            response_format=EventsResponse,
        )
        parsed = completion.choices[0].message.parsed
        print(f"Parsed events: {len(parsed.events)}")
    except Exception as e:
        print("Parse failed:", e)

if __name__ == "__main__":
    asyncio.run(main())

支持的模型

Dedalus SDK 的 .parse() 和 .stream() 方法适用于所有提供商。schema 强制执行方式有所不同： 严格执行（基于 CFG，提供 schema 级别保证）：

✓ openai/* - 上下文无关文法编译
✓ xai/* - 原生 schema 校验
✓ fireworks_ai/* - 原生 schema 校验（部分模型）
✓ deepseek/* - 原生 schema 校验（部分模型）

最佳努力（将 schema 发送给模型作为指导，但不做保证）：

🟡 google/* - 将 schema 转发至 generationConfig.responseSchema
🟡 anthropic/* - 基于 prompt 的 JSON 生成（约 85–90% 成功率）

对于 google/* 和 anthropic/* 模型，请务必验证解析后的输出并实现重试逻辑。

Provider 示例

你可以在任意 Provider 的模型上使用 .parse() 和 .stream()。在实际使用中，你只需要更改 model——其他一切都保持不变。完整的模型 id 列表请参见 providers 指南。

速查

Python (Pydantic)

from dedalus_labs import AsyncDedalus
from pydantic import BaseModel

class MyModel(BaseModel):
    field: str

client = AsyncDedalus()
result = await client.chat.completions.parse(
    model="openai/gpt-5.2",
    messages=[...],
    response_format=MyModel,
)
parsed = result.choices[0].message.parsed

TypeScript（Zod）

import Dedalus from 'dedalus-labs';
import { zodResponseFormat } from 'dedalus-labs/helpers/zod';
import { z } from 'zod';

const MySchema = z.object({ field: z.string() });

const client = new Dedalus();
const result = await client.chat.completions.parse({
  model: 'openai/gpt-5.2',
  messages: [...],
  response_format: zodResponseFormat(MySchema, 'my_schema'),
});
const parsed = result.choices[0]?.message.parsed;

TypeScript（Effect Schema）

import Dedalus from 'dedalus-labs';
import { effectResponseFormat } from 'dedalus-labs/helpers/effect';
import * as Schema from 'effect/Schema';

const MySchema = Schema.Struct({ field: Schema.String });

const client = new Dedalus();
const result = await client.chat.completions.parse({
  model: 'openai/gpt-5.2',
  messages: [...],
  response_format: effectResponseFormat(MySchema, 'my_schema'),
});
const parsed = result.choices[0]?.message.parsed;

Zod 辅助工具

import { zodResponseFormat, zodFunction } from 'dedalus-labs/helpers/zod';

// 用于响应 schema
zodResponseFormat(MyZodSchema, 'schema_name')

// 用于 tool 定义
zodFunction({
  name: 'tool_name',
  description: '该 tool 的功能描述',
  parameters: z.object({ ... }),
  function: (args) => { ... },
})

Effect 辅助函数

import { effectResponseFormat, effectFunction } from 'dedalus-labs/helpers/effect';

// 用于响应 schema
effectResponseFormat(MyEffectSchema, 'schema_name')

// For tool definitions
effectFunction({
  name: 'tool_name',
  description: 'What the tool does',
  parameters: MyEffectParametersSchema,
  function: (args) => { ... },
})

使用 @effect/schema（上游已弃用）

如果你仍在使用 @effect/schema，那么来自 @effect/schema/Schema 的 schema 也可以配合 helpers/effect 使用。你仍然需要安装 effect（Dedalus SDK 使用 effect/JSONSchema 和 effect/Schema 进行转换和验证）。新代码请优先使用 effect/Schema。

后续步骤

流式输出：Streaming — 改善工具/模型上下文协议 (MCP) 长时间运行时的用户体验
跨模型路由：Handoffs — 按阶段搭配使用快速/高性能模型
查看模式：Use Cases — 结构化抽取工作流

以编程方式连接这些文档，通过模型上下文协议 (MCP) 将其接入 Claude、VSCode 等，以获得实时回答。

入门

功能

示例指南

使用场景

指南

更新日志

结构化输出

提取类型化数据

高级用法

Client API（参考）

TypeScript 设置

`.parse()`（非流式传输）

`.stream()`（流式传输）

可选字段

Schema 与模式

嵌套模型

结构化工具调用（高级）

枚举与联合类型

DedalusRunner 应用程序编程接口（API）

.create() vs .parse() vs .stream()

错误处理

支持的模型

Provider 示例

速查

Python (Pydantic)

TypeScript（Zod）

TypeScript（Effect Schema）

Zod 辅助工具

Effect 辅助函数

后续步骤

入门

功能

示例指南

使用场景

指南

更新日志

​提取类型化数据

​高级用法

​Client API（参考）

​TypeScript 设置

​.parse()（非流式传输）

​.stream()（流式传输）

​可选字段

​Schema 与模式

​嵌套模型

​结构化工具调用（高级）

​枚举与联合类型

​DedalusRunner 应用程序编程接口（API）

​.create() vs .parse() vs .stream()

​错误处理

​支持的模型

​Provider 示例

​速查

​Python (Pydantic)

​TypeScript（Zod）

​TypeScript（Effect Schema）

​Zod 辅助工具

​Effect 辅助函数

​后续步骤

提取类型化数据

高级用法

Client API（参考）

TypeScript 设置

`.parse()`（非流式传输）

`.stream()`（流式传输）

可选字段

Schema 与模式

嵌套模型

结构化工具调用（高级）

枚举与联合类型

DedalusRunner 应用程序编程接口（API）

.create() vs .parse() vs .stream()

错误处理

支持的模型

Provider 示例

速查

Python (Pydantic)

TypeScript（Zod）

TypeScript（Effect Schema）

Zod 辅助工具

Effect 辅助函数

后续步骤