智能体开发：为“概率大脑”搭建“反射神经”与“肌肉骨骼”

# 智能体最简生存内核：ReAct 循环骨架
class SimpleAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_iterations = 5  # 工程铁律：必须设定止损阈值

    def run(self, user_goal):
        thought_history = []
        for i in range(self.max_iterations):
            # 1. 推理 (Reasoning): 思考当前状态与下一步行动
            response = self.llm.generate(goal=user_goal, history=thought_history)
            
            # 2. 解析 (Parsing): 从自然语言中提取结构化工具调用
            action, params = self.parse_action(response)
            
            # 3. 执行 (Acting): 调用外部工具改变世界状态
            observation = self.tools[action].execute(**params)
            
            # 4. 观察与记录 (Observation): 将反馈压入记忆，作为下一轮推理依据
            thought_history.append({"action": action, "result": observation})
            
            # 5. 终止判断: 若智能体认为目标已达成或触发异常
            if self.is_goal_achieved(observation):
                return observation
        return "达到最大迭代次数，任务终止。"

from pydantic import BaseModel, Field
from typing import Literal

# 定义“工具”的刚性契约（智能体侧的接口定义）
class FileOperation(BaseModel):
    action_type: Literal["read", "write", "delete"] = Field(..., description="操作类型")
    file_path: str = Field(..., regex=r"^/data/.*\.txt$", description="必须为绝对路径")
    content: str = Field(None, description="写入时的内容，读取删除时可为空")

# 解析器中的强制转换逻辑
def safe_execute(raw_llm_output: str):
    try:
        # 若 LLM 输出的 JSON 不合规，此处直接抛出异常，绝不往下传递脏数据
        cmd = FileOperation.model_validate_json(raw_llm_output)
        return os_interface(cmd.action_type, cmd.file_path, cmd.content)
    except ValidationError:
        return "指令格式错误，已拒绝执行。请重新表述。"

class MemoryManager:
    def __init__(self, short_term_limit=4000):
        self.short_term = []  # 当前轮次的“工作台”
        self.long_term = []   # 向量化存储的“档案馆”

    def add_observation(self, obs):
        self.short_term.append(obs)
        # 工程红线：一旦短期记忆超过 4000 字符
        if self.count_tokens(self.short_term) > 4000:
            # 强制将前半部分压缩为 100 字的摘要，扔进长期库
            summary = self.llm.summarize(self.short_term[:-3]) 
            self.long_term.append(summary)
            # 仅保留最近 3 轮交互作为当前上下文
            self.short_term = self.short_term[-3:] 

import json
from datetime import datetime

def log_thought_cycle(cycle_num, thought, action, observation):
    # 将每一步思考、动作、观察，以 JSONL 格式写入持久化存储
    entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "cycle": cycle_num,
        "thought_preview": thought[:50],
        "action": action,
        "observation_hash": hash(observation)  # 便于去重与回溯
    }
    with open(f"./agent_logs/{datetime.now().date()}.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")
    
    # 同时在控制台输出带颜色的状态，让开发者实时感知
    print(f"\033[92m[Cycle {cycle_num}]\033[0m Action: {action}")