🐍 Day06 Practice Template

📄 文件: day06_practice_template.py
"""
Day 6 练习：不依赖框架，纯 Python 手写一个 ReAct Agent 循环
验收标准：
1. 必须实现 ReAct 循环（Reasoning + Acting 交替）
2. 循环必须有 max_steps 上限（防止无限循环）
3. 必须支持工具调用（至少 2 个工具）
4. 必须使用 async/await（异步 LLM 调用）
5. 必须正确解析 LLM 的 Thought/Action/Action Input/Observation 格式
6. 最终输出 Final Answer
"""

import asyncio
import json
import re
import functools
from typing import List, Dict, Any, Optional, Callable, Tuple
from dataclasses import dataclass, field


# ==================== 数据结构 ====================

@dataclass
class ToolResult:
    """工具调用结果"""
    tool_name: str
    tool_input: str
    output: str
    success: bool


@dataclass
class AgentStep:
    """Agent 单步执行记录"""
    thought: str = ""
    action: str = ""
    action_input: str = ""
    observation: str = ""
    full_response: str = ""


# ==================== 任务 1：实现工具注册和调用 ====================

# 全局工具注册表
TOOLS_REGISTRY: Dict[str, Dict[str, Any]] = {}


def tool(name: str, description: str):
    """工具注册装饰器"""
    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        TOOLS_REGISTRY[name] = {
            "function": wrapper,
            "description": description,
            "parameters": func.__code__.co_varnames[:func.__code__.co_argcount]
        }
        return wrapper
    return decorator


# TODO 1.1：注册至少 2 个工具
# 提示：参考 Day 2 的写法，定义计算器工具和天气查询工具

@tool("Calculator", "执行四则运算，例如：Calculator[3 + 5 * 2]")
def calculator(expression: str) -> str:
    """计算器工具"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"错误: {e}"


@tool("Weather", "查询城市天气，例如：Weather[北京]")
def weather(city: str) -> str:
    """天气查询工具（模拟）"""
    # 模拟天气数据
    天气数据 = {
        "北京": "晴，25°C",
        "上海": "多云，22°C",
        "深圳": "雨，28°C",
    }
    return 天气数据.get(city, f"{city} 的天气数据未找到")


def get_tools_description() -> str:
    """生成工具描述文本（放入 System Prompt）"""
    # TODO 1.2：实现工具描述生成
    # 格式：
    # Calculator[参数]: 描述
    # Weather[参数]: 描述
    pass


def execute_tool(tool_name: str, tool_input: str) -> ToolResult:
    """执行工具调用"""
    # TODO 1.3：实现工具执行逻辑
    # 1. 从 TOOLS_REGISTRY 查找工具
    # 2. 调用工具的 "function"
    # 3. 返回 ToolResult
    pass


# ==================== 任务 2：实现 ReAct Prompt 构建 ====================

def build_react_prompt(user_input: str, steps: List[AgentStep]) -> str:
    """
    构建 ReAct 格式的 Prompt
    
    ReAct 格式：
    Question: [用户问题]
    
    Thought: [推理过程]
    Action: [工具名]
    Action Input: [工具输入]
    Observation: [工具返回结果]
    
    ...（可重复多轮）
    
    Thought: [最终推理]
    Final Answer: [最终答案]
    """
    # TODO 2.1：实现 ReAct Prompt 构建
    # 提示：
    # 1. System Prompt 包含工具描述和 ReAct 格式说明
    # 2. 把历史 steps 格式化为 Thought/Action/Observation 序列
    # 3. 最后加上当前问题
    pass


# ==================== 任务 3：实现 ReAct 格式解析 ====================

def parse_react_response(response: str) -> Tuple[str, str, str]:
    """
    解析 LLM 返回的 ReAct 格式文本
    
    返回：(thought, action, action_input)
    - 如果包含 "Final Answer:"，说明是最终回答，action 为空
    - 否则解析 Action 和 Action Input
    
    示例响应：
    Thought: 我需要计算 3 + 5 的值
    Action: Calculator
    Action Input: 3 + 5 * 2
    """
    # TODO 3.1：实现 ReAct 响应解析
    # 提示：
    # 1. 用正则或字符串查找提取 Thought、Action、Action Input
    # 2. 检查是否包含 "Final Answer"（区分大小写）
    # 3. 返回 (thought, action, action_input)
    #    如果 action 为空，说明是最终回答，此时 action_input 存最终答案
    pass


# ==================== 任务 4：实现 ReAct Agent 主循环 ====================

async def react_agent(
    user_input: str,
    max_steps: int = 10,
    verbose: bool = True,
) -> str:
    """
    纯 Python 实现的 ReAct Agent 主循环
    
    流程：
    1. 初始化 steps = []
    2. for step in range(max_steps):
    3.   构建 prompt（build_react_prompt）
    4.   调用 LLM（模拟或真实 API）
    5.   解析响应（parse_react_response）
    6.   如果是 Final Answer → 返回结果，结束
    7.   否则执行工具（execute_tool）
    8.   把结果加入 steps，继续循环
    9. 如果超过 max_steps → 返回错误信息
    """
    steps: List[AgentStep] = []
    
    # TODO 4.1：实现 ReAct Agent 主循环
    # 提示：这里是整个 Agent 的核心！
    # 注意：LLM 调用可以用模拟函数（见下方 mock_llm_call）
    
    pass


# ==================== 模拟 LLM（用于测试，不依赖真实 API） ====================

async def mock_llm_call(prompt: str) -> str:
    """
    模拟 LLM 调用（用于本地测试，不消耗 API 额度）
    
    根据 prompt 的内容，返回一个符合 ReAct 格式的模拟响应。
    判定规则：
    1. 找 prompt 中最后一个真实的 Action 行（忽略格式说明里的占位行）；
    2. 如果该 Action 后面已经跟了 Observation，说明工具已执行过，
       返回引用工具结果的 Final Answer → 循环正常终止；
    3. 否则返回对应的 Action，让 Agent 去执行工具。
    
    因此只要你的 react_agent 按 ReAct 格式把历史步骤拼进 prompt，
    第 2 轮就会拿到 Final Answer，不会死循环。
    """
    # 1. 找最后一个真实 Action（排除格式说明中的占位行，如 "Action: [工具名]"）
    real_actions = [
        m for m in re.finditer(r"Action:\s*(.+)", prompt)
        if "[" not in m.group(1) and "工具名" not in m.group(1)
    ]
    
    if real_actions:
        last = real_actions[-1]
        after_action = prompt[last.end():]
        if "Observation:" in after_action:
            obs_match = re.search(r"Observation:\s*(.+)", after_action)
            observation = obs_match.group(1).strip() if obs_match else ""
            return f"""Thought: 我已经通过工具调用获得了所需信息
Final Answer: 根据工具返回的结果（{observation}），问题已解决。"""
    
    # 2. 还没执行过工具 → 返回 Action 让 Agent 执行
    #    关键词只匹配用户的 Question 行，避免被工具描述里的示例（如 "3 + 5 * 2"）干扰
    question_match = re.search(r"Question:\s*(.+)$", prompt, re.MULTILINE)
    判定文本 = question_match.group(1) if question_match else prompt

    if "计算" in 判定文本 or "加" in 判定文本 or "减" in 判定文本 or "+" in 判定文本:
        return """Thought: 我需要计算这个表达式的结果
Action: Calculator
Action Input: 3 + 5 * 2"""
    
    if "天气" in 判定文本:
        return """Thought: 我需要查询北京的天气
Action: Weather
Action Input: 北京"""
    
    # 3. 不需要工具的问题，直接回答
    return """Thought: 这个问题不需要使用工具
Final Answer: 我是直接根据已知知识回答的。"""


# ==================== 任务 5：用真实 LLM API 替换 mock ====================
# （可选，有 API key 再做；先用 mock 把逻辑跑通）

async def real_llm_call(prompt: str, api_key: str = "") -> str:
    """
    真实 LLM API 调用（OpenAI 兼容格式）
    
    支持：DeepSeek、Qwen、Moonshot 等
    """
    # TODO 5.1：实现真实 LLM 调用
    # 提示：
    # 1. 用 aiohttp 或 httpx 发送异步 HTTP 请求
    # 2. API 格式兼容 OpenAI：
    #    POST https://api.deepseek.com/chat/completions
    #    Headers: {"Authorization": f"Bearer {api_key}"}
    #    Body: {"model": "...", "messages": [{"role": "user", "content": prompt}]}
    # 3. 返回 response.json()["choices"][0]["message"]["content"]
    pass


# ==================== 主函数：测试 Agent ====================

async def main():
    print("=" * 60)
    print("Day 6 练习：纯 Python 手写 ReAct Agent")
    print("=" * 60)
    
    # 测试用例 1：计算问题
    print("\n📝 测试 1：计算问题")
    print("用户：帮我计算 3 + 5 * 2 等于多少？")
    
    答案 = await react_agent(
        user_input="帮我计算 3 + 5 * 2 等于多少？",
        max_steps=5,
        verbose=True,
    )
    print(f"\n✅ Agent 最终回答：{答案}")
    
    # 测试用例 2：天气问题
    print("\n\n📝 测试 2：天气查询")
    print("用户：北京今天天气怎么样？")
    
    答案 = await react_agent(
        user_input="北京今天天气怎么样？",
        max_steps=5,
        verbose=True,
    )
    print(f"\n✅ Agent 最终回答：{答案}")
    
    print("\n" + "=" * 60)
    print("✅ 所有测试完成！现在运行 python day06_practice_validator.py 验收")
    print("=" * 60)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"\n⚠️ 运行中断：{type(e).__name__}: {e}")
        print("💡 这通常是因为上面的 TODO 还没有完成。请先实现各个任务函数，再运行本文件。")
资源导航

实战教程

Day 1-3 (基础)

Day 4-5 (LLM)

Day 6-7 (Agent)

Day 8-10 (高级)

Day 11-15 (项目实战)

🐍 Day06 Practice Template
