Prompt Engineering 演进：从 Zero-shot 到 ReAct

prompt_zero_shot = """
将以下文本分类为正面或负。

文本：这家餐厅的服务太棒了，菜品也很好吃。
情感：
"""
# 输出：正面

prompt_few_shot = """
将以下文本分类为正面或负。

文本：这家餐厅的服务太棒了
情感：正面

文本：等位等了 2 小时，太糟糕了
情感：负

文本：菜味道一般，没什么特别
情感：
"""
# 输出：中性

prompt_cot = """
问：Roger 一开始有 5 个网球。他又买了 2 罐网球，每罐 3 个。他现在有多少个？
思考：Roger 一开始有 5 个球。2 罐每罐 3 个 = 6 个球。5 + 6 = 11。答案是 11。

问：食堂有 23 个苹果。用掉 20 个做午餐，又买了 6 个。现在有多少个？
思考：
"""
# 模型输出：食堂有 23 个苹果。用掉 20 个剩 3 个。又买了 6 个 = 9 个。答案是 9。

prompt_zero_shot_cot = """
问：食堂有 23 个苹果。用掉 20 个做午餐，又买了 6 个。现在有多少个？
让我们一步步思考。
"""

REACT_PROMPT = """
回答以下问题，你可以使用以下工具：
- Search[query]：搜索网页
- Lookup[term]：在当前文档中查找术语
- Finish[answer]：结束并给出答案

按照以下格式：
Thought: 思考下一步行动
Action: 选择的工具和参数
Observation: 工具返回结果
... (重复 Thought/Action/Observation)
Thought: 我现在知道答案了
Action: Finish[最终答案]

问题：{question}
{scratchpad}
"""

def reflexion_loop(task, max_trials=3):
    memory = []
    for trial in range(max_trials):
        # 1. 基于历史反思 + 当前任务生成 action
        action = agent.act(task, memory)
        
        # 2. 执行并观察
        observation = env.step(action)
        
        # 3. 判断是否成功
        if is_success(observation):
            return observation
        
        # 4. 反思：为什么失败？
        reflection = llm.generate(
            f"任务：{task}\n"
            f"尝试：{action}\n"
            f"结果：{observation}\n"
            f"反思："
        )
        memory.append(reflection)
    
    return failure

class TreeOfThoughts:
    def __init__(self):
        self.thought_generator = llm.generate_k_candidates  # k 个候选
        self.state_evaluator = llm.evaluate_state  # 给状态打分
        self.search_algorithm = BFS or DFS  # 搜索策略
    
    def solve(self, problem):
        root = Node(state=problem, thought="")
        frontier = [root]
        while frontier:
            node = self.search_algorithm.pop(frontier)
            # 生成候选思路
            candidates = self.thought_generator(node.state, k=5)
            for candidate in candidates:
                new_node = Node(state=apply(node.state, candidate), thought=candidate)
                score = self.state_evaluator(new_node.state)
                if score > 0.8:
                    return new_node.state
                if score > 0.5:
                    frontier.append(new_node)
        return None

# 传统手工 prompt
prompt = """
你是一个专业的金融分析师。请分析以下公司的财报：
{report}
按照以下 JSON 格式输出：
{{"summary": "...", "risk": "...", "rating": "buy|hold|sell"}}
"""

# DSPy 风格
import dspy

class FinancialAnalyst(dspy.Signature):
    """分析公司财报并给出投资评级"""
    report: str = dspy.InputField()
    summary: str = dspy.OutputField()
    risk: str = dspy.OutputField()
    rating: str = dspy.OutputField()

class AnalystAgent(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze = dspy.ChainOfThought(FinancialAnalyst)
    
    def forward(self, report):
        return self.analyze(report=report)

# 自动编译：DSPy 用 few-shot 示例 + teleprompter 自动找到最佳 prompt
agent = AnalystAgent()
optimizer = dspy.BootstrapFewShot(metric=accuracy_metric)
optimized_agent = optimizer.compile(agent, trainset=trainset)

import dspy

class ReactAgent(dspy.Module):
    def __init__(self, tools):
        super().__init__()
        self.react = dspy.ReAct(
            signature="question -> answer",
            tools=tools,  # [search, calculator, lookup]
            max_iters=5,
        )
        self.reflector = dspy.ChainOfThought(
            "task, trajectory, outcome -> reflection"
        )
    
    def forward(self, question, max_trials=3):
        for trial in range(max_trials):
            prediction = self.react(question=question)
            if self.is_success(prediction):
                return prediction
            
            # 反思失败原因
            reflection = self.reflector(
                task=question,
                trajectory=prediction.trajectory,
                outcome=prediction.answer,
            )
            
            # 把反思加入下一次的 context
            question = f"{question}\n\nPrevious reflection: {reflection.reflection}"
        
        return prediction

# DSPy 自动编译这个组合 Agent 的所有 prompt
optimized = dspy.MIPROv2(metric=success_rate).compile(
    ReactAgent(tools=[search, calculator]),
    trainset=trainset,
)

import langsmith

@langsmith.traceable
def react_agent(question):
    trajectory = []
    for step in range(max_steps):
        thought = llm.generate(f"Question: {question}\nTrajectory: {trajectory}\nThought:")
        action = parse_action(thought)
        observation = tools[action.name](*action.args)
        trajectory.append({"thought": thought, "action": action, "observation": observation})
        if action.name == "Finish":
            return action.args[0]
    return None