内置 Module:Predict / ChainOfThought / ReAct

四大内置 Module 速查

Predict:最直接的调用

class Categorize(dspy.Signature):
    """把邮件归类为三个类别之一"""
    email: str = dspy.InputField()
    category: Literal["spam", "work", "personal"] = dspy.OutputField()

cat = dspy.Predict(Categorize)
cat(email="您的订单已发货...").category    # → "work"

Predict 是最纯粹的 Signature 执行器:拼 prompt → 调 LLM → 解析输出。适合不需要思考过程的快速任务。

ChainOfThought:加一个 reasoning 字段

cot = dspy.ChainOfThought("question -> answer")
result = cot(question="小明有 5 个苹果,分给 3 个朋友,每人至少 1 个,最多 2 个,有多少种分法?")
print(result.reasoning)   # 步骤推理
print(result.answer)      # 答案

底层等价于一个新的 Signature:

# ChainOfThought 其实做的事
class QAWithReasoning(dspy.Signature):
    """..."""
    question: str = dspy.InputField()
    reasoning: str = dspy.OutputField(desc="分步思考")
    answer: str = dspy.OutputField()

和 reasoning 字段相关的技巧

# 自定义 rationale 字段名和描述
cot = dspy.ChainOfThought(
    QA,
    rationale_type=dspy.OutputField(
        prefix="分析过程",
        desc="从给定 context 中找证据并逐步推导",
    ),
)

ReAct:工具调用循环

Reason-Act-Observe 的经典模式。DSPy 2.5 里的 ReAct 已经很接近生产可用:

def search_web(query: str) -> str:
    """搜索引擎,返回前 3 条摘要。"""
    return tavily.search(query)

def calc(expr: str) -> str:
    """计算数学表达式,如 '2+2*3'。"""
    return str(eval(expr))   # 生产请用 sympy

agent = dspy.ReAct(
    "question -> answer",
    tools=[search_web, calc],
    max_iters=5,
)

out = agent(question="马斯克现年多少岁?再加上圆周率前三位之和。")

DSPy 会自动:

1. 从工具的 docstring 和类型提示生成工具签名,塞进 prompt
2. 解析 LLM 输出的 Thought: ... / Action: tool(args) / Observation:
3. 执行工具,把返回拼进下一轮 context
4. 发现 "Finish" 或达到 max_iters 就停

查看 trajectory

out = agent(question="...")
print(out.trajectory)    # 完整的 Thought/Action/Observation 序列

ProgramOfThought:让 LLM 写代码算

pot = dspy.ProgramOfThought("question -> answer")
pot(question="2024 年有几个周三?").answer

LLM 生成一段 Python,DSPy 在沙箱里跑,把结果塞回去。对于:

• 精确的数学运算(LLM 算数学普遍拉胯)
• 日期/时间计算
• 需要确定性逻辑的统计

ProgramOfThought 比 CoT 靠谱得多——因为 Python 不会"幻觉"。

Module 组合优先于继承

上面四个都是"基础元件",复杂逻辑通过把它们塞到 dspy.Module 的 forward 里组合:

class EmailAssistant(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict("email -> category: Literal['spam','work','personal']")
        self.summarize = dspy.ChainOfThought("email -> summary, action_items: list[str]")

    def forward(self, email):
        cat = self.classify(email=email).category
        if cat == "spam":
            return dspy.Prediction(category=cat, summary="(已标记垃圾)", action_items=[])
        out = self.summarize(email=email)
        return dspy.Prediction(category=cat, summary=out.summary, action_items=out.action_items)

选择哪个 Module 的决策树

控制解码参数

# 单次配置
out = cot(question=q, config={"temperature": 0.2, "max_tokens": 512})

# 全局默认
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini", temperature=0, max_tokens=1000))

N-best 采样

outs = cot.forward(question=q, n=5)    # 返回 5 个候选
# 配合自定义投票/评分挑最佳

常见误区

本章小结