Skip to main content
策略允许在智能体执行的每一步中注入逻辑。可以根据运行时上下文（如步骤数量、先前的输出或外部状态）添加指令、修改行为并强制执行约束条件。

​
基本策略

策略是一个接收上下文并返回修改结果的函数： 
import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dotenv import load_dotenv

load_dotenv()

def policy(ctx: dict) -> dict:
    step = ctx.get("step", 1)
    
    if step >= 3:
        # 第 3 步之后，提示模型收尾作答
        return {
            "message_prepend": [
                {"role": "system", "content": "Provide your final answer now."}
            ],
            "max_steps": 4
        }
    
    return {}

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

    result = runner.run(
        input="Research the history of the internet and summarize key milestones",
        model="openai/gpt-5.2",
        mcp_servers=["windsor/brave-search-mcp"],  # 通过 Brave Search 模型上下文协议 (MCP) 进行网页搜索
        stream=True,
        policy=policy
    )

    await stream_async(result)

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

​
策略上下文

ctx 字典包含：
字段类型描述
stepint当前执行步骤（从 1 开始计数）
messageslist到目前为止的会话历史
tools_calledlist在之前步骤中调用的工具

​
策略返回值

策略可以返回：
字段作用
message_prepend / messagePrepend在下一次模型调用之前追加的 messages
message_append / messageAppend在对话之后追加的 messages
max_steps / maxSteps覆盖最大步数限制
stop表示是否提前停止执行的布尔值

​
使用场景

限流：跨多个步骤跟踪应用程序编程接口 API 调用，在接近限制时暂停。 安全护栏：检查输出是否存在策略违规，并注入用于纠正的 prompt。 动态指令：根据中间结果调整行为。 成本控制：在执行了一定数量的高成本操作后停止执行。

​
工具事件回调

使用 on_tool_event 监控 tool 的执行情况： 
import json

def on_tool(evt: dict) -> None:
    print(f"Tool called: {json.dumps(evt)}")

result = runner.run(
    input="为一个 5kg 的包裹计算寄往伦敦的运费",
    model="openai/gpt-5.2",
    tools=[calculate_shipping],
    on_tool_event=on_tool,
    policy=policy
)

​
后续步骤

  • 工具 — 定义策略可控制的工具
通过模型上下文协议（MCP）将以编程方式连接这些文档到 Claude、VSCode 等，以获得实时解答。