# ai_agent_minimal.py
# 实现 Agent 最小循环：感知 → 推理 → 执行 → 反馈
import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "sk-fake"))

SYSTEM_PROMPT = """你是 ReAct Agent。每一步必须输出 JSON：
{"thought": "当前思考", "action": {"tool": "read_file", "args": {"path": "..."}}}
若已获得足够信息，输出：{"thought": "...", "final_answer": "..."}"""

def read_file(path: str) -> str:
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"[error] {e}"

def run_agent(task: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": task},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={"type": "json_object"},
        )
        content = resp.choices[0].message.content
        parsed = json.loads(content)
        print(f"[step {step + 1}] {parsed.get('thought', '')}")

        if "final_answer" in parsed:
            return parsed["final_answer"]

        action = parsed.get("action", {})
        if action.get("tool") == "read_file":
            result = read_file(action["args"]["path"])
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"[observation] {result}"})

    return "[timeout] 未能在限定步数内完成"

if __name__ == "__main__":
    # 运行前请在当前目录创建 workspace/note.txt
    print(run_agent("请读取 ./workspace/note.txt 并总结内容"))

# ai_agent_state_machine.py
# 用状态机约束 Agent 活动路径，防止乱跑
from enum import Enum, auto

class AgentState(Enum):
    IDLE = auto()
    PERCEIVE = auto()
    REASON = auto()
    EXECUTE = auto()
    FEEDBACK = auto()
    DONE = auto()

class GuardedAgent:
    TRANSITIONS = {
        AgentState.IDLE: [AgentState.PERCEIVE],
        AgentState.PERCEIVE: [AgentState.REASON],
        AgentState.REASON: [AgentState.EXECUTE, AgentState.DONE],
        AgentState.EXECUTE: [AgentState.FEEDBACK],
        AgentState.FEEDBACK: [AgentState.PERCEIVE],
    }

    def __init__(self):
        self.state = AgentState.IDLE
        self.context = []

    def transit(self, target: AgentState):
        if target not in self.TRANSITIONS[self.state]:
            raise ValueError(f"非法状态转移: {self.state.name} -> {target.name}")
        self.state = target
        print(f"state -> {target.name}")

    def run(self, task: str, max_steps: int = 5):
        self.transit(AgentState.PERCEIVE)
        self.context.append(task)
        for _ in range(max_steps):
            self.transit(AgentState.REASON)
            action = "read_file" if "read" in task else "search"
            self.transit(AgentState.EXECUTE)
            obs = f"完成 {action}"
            self.transit(AgentState.FEEDBACK)
            self.context.append(obs)
            if len(self.context) > 3:
                self.transit(AgentState.DONE)
                return "done"
        return "timeout"

if __name__ == "__main__":
    agent = GuardedAgent()
    print(agent.run("读取配置文件并分析"))

// ai_agent_minimal.go
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
)

const openAIURL = "https://api.openai.com/v1/chat/completions"

var systemPrompt = `你是 ReAct Agent。每一步必须输出 JSON：
{"thought": "...", "action": {"tool": "read_file", "args": {"path": "..."}}}
若已获得足够信息，输出：{"thought": "...", "final_answer": "..."}`

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

func readFile(path string) string {
	b, err := os.ReadFile(path)
	if err != nil {
		return fmt.Sprintf("[error] %v", err)
	}
	return string(b)
}

func chat(messages []Message) (map[string]interface{}, error) {
	body, _ := json.Marshal(map[string]interface{}{
		"model":           "gpt-4o-mini",
		"messages":        messages,
		"response_format": map[string]string{"type": "json_object"},
	})
	req, _ := http.NewRequest("POST", openAIURL, bytes.NewReader(body))
	req.Header.Set("Authorization", "Bearer "+os.Getenv("OPENAI_API_KEY"))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	data, _ := io.ReadAll(resp.Body)
	var result map[string]interface{}
	json.Unmarshal(data, &result)
	return result, nil
}

func runAgent(task string, maxSteps int) string {
	messages := []Message{
		{Role: "system", Content: systemPrompt},
		{Role: "user", Content: task},
	}
	for step := 0; step < maxSteps; step++ {
		resp, err := chat(messages)
		if err != nil {
			return "[error] " + err.Error()
		}
		content := resp["choices"].([]interface{})[0].(map[string]interface{})["message"].(map[string]interface{})["content"].(string)
		var parsed map[string]interface{}
		json.Unmarshal([]byte(content), &parsed)
		fmt.Printf("[step %d] %s\n", step+1, parsed["thought"])
		if final, ok := parsed["final_answer"]; ok {
			return final.(string)
		}
		action := parsed["action"].(map[string]interface{})
		if action["tool"] == "read_file" {
			args := action["args"].(map[string]interface{})
			result := readFile(args["path"].(string))
			messages = append(messages, Message{Role: "assistant", Content: content})
			messages = append(messages, Message{Role: "user", Content: "[observation] " + result})
		}
	}
	return "[timeout] 未能在限定步数内完成"
}

func main() {
	fmt.Println(runAgent("请读取 ./workspace/note.txt 并总结内容", 5))
}

# task_planner.py
# 用 Plan / Step / Type 三种模式执行同一研究任务
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Callable

class StepType(Enum):
    WEBSEARCH = "websearch"
    RESEARCH = "research"
    PROCESSING = "processing"

@dataclass
class Step:
    id: str
    step_type: StepType
    description: str
    status: str = "todo"
    result: str = ""

@dataclass
class Plan:
    goal: str
    steps: List[Step] = field(default_factory=list)

class TaskPlanner:
    def __init__(self):
        self.handlers: dict[StepType, Callable[[Step], str]] = {
            StepType.WEBSEARCH: lambda s: f"搜索完成：{s.description}",
            StepType.RESEARCH: lambda s: f"研究完成：{s.description}",
            StepType.PROCESSING: lambda s: f"汇总完成：{s.description}",
        }

    def create_plan(self, goal: str) -> Plan:
        return Plan(goal=goal, steps=[
            Step("s1", StepType.WEBSEARCH, f"搜索 {goal} 的背景信息"),
            Step("s2", StepType.RESEARCH, f"深度分析 {goal}"),
            Step("s3", StepType.PROCESSING, "汇总生成报告"),
        ])

    def run_react(self, goal: str, max_steps: int = 5) -> str:
        steps = []
        for i in range(max_steps):
            thought = f"Step {i+1}: 继续推进目标"
            action = "websearch" if i == 0 else ("research" if i == 1 else "report")
            obs = f"观察到 {action} 的结果"
            steps.append(f"{thought} -> {action} -> {obs}")
            if action == "report":
                break
        return "\n".join(steps)

    def run_plan_execute(self, plan: Plan) -> str:
        for step in plan.steps:
            step.status = "running"
            step.result = self.handlers[step.step_type](step)
            step.status = "done"
        return self._report(plan)

    def run_typed(self, plan: Plan, max_parallel: int = 2) -> str:
        while any(s.status in ("todo", "running") for s in plan.steps):
            batch = [s for s in plan.steps if s.status == "todo"][:max_parallel]
            for step in batch:
                step.status = "running"
                step.result = self.handlers[step.step_type](step)
                step.status = "done"
        return self._report(plan)

    def _report(self, plan: Plan) -> str:
        lines = [f"# {plan.goal}", ""]
        for s in plan.steps:
            lines.append(f"- [{s.status}] {s.result}")
        return "\n".join(lines)

if __name__ == "__main__":
    planner = TaskPlanner()
    goal = "2025 年国内 AI Agent 市场格局"
    plan = planner.create_plan(goal)
    print("=== ReAct ===")
    print(planner.run_react(goal))
    print("\n=== Plan-and-Execute ===")
    print(planner.run_plan_execute(plan))
    print("\n=== Plan/Step/Type ===")
    print(planner.run_typed(planner.create_plan(goal)))

# replan_trigger.py
# 当 Step 失败或观察到新信息时触发重新规划
import json

def execute_step(step: dict) -> dict:
    # 模拟执行，90% 成功
    import random
    if random.random() < 0.9:
        return {"status": "done", "output": f"完成 {step['name']}"}
    return {"status": "failed", "error": "依赖未就绪"}

def plan_and_execute(goal: str, initial_plan: list, max_replan: int = 3):
    plan = initial_plan
    context = {"goal": goal, "observations": []}
    for attempt in range(max_replan + 1):
        print(f"\n[plan attempt {attempt + 1}] plan={json.dumps(plan, ensure_ascii=False)}")
        for step in plan[:]:
            result = execute_step(step)
            context["observations"].append({step["name"]: result})
            if result["status"] == "failed":
                print(f"  step {step['name']} failed, trigger replan")
                plan = replan(context)
                break
        else:
            return f"成功完成：{goal}"
    return f"失败：{goal}"

def replan(context: dict) -> list:
    # 简化版：把未完成的步骤和新增补偿步骤加入新计划
    remaining = [s for s in context["observations"][-3:] if isinstance(s, dict) and list(s.values())[0].get("status") == "failed"]
    new_plan = [{"name": "fix_dependency"}, {"name": "retry_main_task"}]
    return new_plan

if __name__ == "__main__":
    print(plan_and_execute("部署服务", [{"name": "build"}, {"name": "deploy"}]))

// TaskPlanner.java
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

class TaskPlanner {
    enum StepType { WEBSEARCH, RESEARCH, PROCESSING }

    static class Step {
        String id;
        StepType type;
        String description;
        String status = "todo";
        String result = "";
        Step(String id, StepType type, String description) {
            this.id = id; this.type = type; this.description = description;
        }
    }

    static class Plan {
        String goal;
        List<Step> steps = new ArrayList<>();
    }

    private final Map<StepType, Function<Step, String>> handlers = Map.of(
        StepType.WEBSEARCH, s -> "搜索完成：" + s.description,
        StepType.RESEARCH, s -> "研究完成：" + s.description,
        StepType.PROCESSING, s -> "汇总完成：" + s.description
    );

    Plan createPlan(String goal) {
        Plan plan = new Plan();
        plan.goal = goal;
        plan.steps.add(new Step("s1", StepType.WEBSEARCH, "搜索 " + goal + " 的背景信息"));
        plan.steps.add(new Step("s2", StepType.RESEARCH, "深度分析 " + goal));
        plan.steps.add(new Step("s3", StepType.PROCESSING, "汇总生成报告"));
        return plan;
    }

    String runPlanExecute(Plan plan) {
        for (Step step : plan.steps) {
            step.status = "running";
            step.result = handlers.get(step.type).apply(step);
            step.status = "done";
        }
        return report(plan);
    }

    String report(Plan plan) {
        StringBuilder sb = new StringBuilder("# ").append(plan.goal).append("\n\n");
        for (Step s : plan.steps) sb.append("- [").append(s.status).append("] ").append(s.result).append("\n");
        return sb.toString();
    }

    public static void main(String[] args) {
        TaskPlanner planner = new TaskPlanner();
        System.out.println(planner.runPlanExecute(planner.createPlan("2025 年国内 AI Agent 市场格局")));
    }
}

# orchestrator_worker.py
# Orchestrator 按步骤调度多个 Worker 完成复合任务
import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "sk-fake-key"))

def query_weather(city: str) -> dict:
    return {"city": city, "temperature": 28, "condition": "sunny"}

def send_email(to: str, subject: str, body: str) -> dict:
    return {"success": True, "message_id": f"msg-{hash(to + subject) & 0xFFFFFF:06x}"}

WORKERS = {
    "query_weather": query_weather,
    "send_email": send_email,
}

SYSTEM_PROMPT = """你是 Orchestrator。用户请求可能需要多个 Worker 协作。
每一步输出 JSON：
{"thought": "...", "step": {"worker": "query_weather", "args": {"city": "..."}}}
若已完成所有步骤，输出：{"thought": "...", "final_answer": "..."}"""

def orchestrate(task: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": task},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={"type": "json_object"},
        )
        content = resp.choices[0].message.content
        parsed = json.loads(content)
        print(f"[step {step + 1}] {parsed.get('thought', '')}")

        if "final_answer" in parsed:
            return parsed["final_answer"]

        step_plan = parsed.get("step", {})
        worker_name = step_plan.get("worker")
        args = step_plan.get("args", {})
        fn = WORKERS.get(worker_name)
        try:
            result = fn(**args) if fn else {"error": f"unknown worker {worker_name}"}
        except Exception as e:
            result = {"error": str(e)}

        messages.append({"role": "assistant", "content": content})
        messages.append({"role": "user", "content": f"[observation] {json.dumps(result, ensure_ascii=False)}"})

    return "[timeout] 未能在限定步数内完成"

if __name__ == "__main__":
    answer = orchestrate("查询北京天气，并将结果通过邮件发送给 bob@example.com")
    print("\n[final]", answer)

# cross_review.py
# 双 Agent 并行审查，互相校验结果
def agent_a_generate(requirement: str) -> str:
    return f"方案 A：基于 {requirement} 设计 REST API"

def agent_b_review(proposal: str) -> dict:
    issues = []
    if "鉴权" not in proposal:
        issues.append("缺少鉴权设计")
    if "限流" not in proposal:
        issues.append("缺少限流设计")
    return {"approved": len(issues) == 0, "issues": issues}

def agent_c_revise(proposal: str, review: dict) -> str:
    if review["approved"]:
        return proposal
    return proposal + "\n补充：增加 JWT 鉴权 + 令牌桶限流"

def cross_review(requirement: str) -> str:
    proposal = agent_a_generate(requirement)
    review = agent_b_review(proposal)
    final = agent_c_revise(proposal, review)
    return final

if __name__ == "__main__":
    print(cross_review("设计用户登录接口"))

// orchestrator_worker.go
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
)

const openAIURL = "https://api.openai.com/v1/chat/completions"

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

func queryWeather(city string) map[string]interface{} {
	return map[string]interface{}{"city": city, "temperature": 28, "condition": "sunny"}
}

func sendEmail(to, subject, body string) map[string]interface{} {
	return map[string]interface{}{"success": true, "message_id": fmt.Sprintf("msg-%d", hashStr(to+subject)&0xFFFFFF)}
}

func hashStr(s string) int {
	h := 0
	for _, c := range s {
		h = 31*h + int(c)
	}
	return h
}

var workers = map[string]func(map[string]interface{}) map[string]interface{}{
	"query_weather": func(args map[string]interface{}) map[string]interface{} {
		return queryWeather(args["city"].(string))
	},
	"send_email": func(args map[string]interface{}) map[string]interface{} {
		return sendEmail(args["to"].(string), args["subject"].(string), args["body"].(string))
	},
}

var systemPrompt = `你是 Orchestrator。用户请求可能需要多个 Worker 协作。
每一步输出 JSON：
{"thought": "...", "step": {"worker": "query_weather", "args": {"city": "..."}}}
若已完成所有步骤，输出：{"thought": "...", "final_answer": "..."}`

func chat(messages []Message) (map[string]interface{}, error) {
	body, _ := json.Marshal(map[string]interface{}{
		"model": "gpt-4o-mini", "messages": messages,
		"response_format": map[string]string{"type": "json_object"},
	})
	req, _ := http.NewRequest("POST", openAIURL, bytes.NewReader(body))
	req.Header.Set("Authorization", "Bearer "+os.Getenv("OPENAI_API_KEY"))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	data, _ := io.ReadAll(resp.Body)
	var result map[string]interface{}
	json.Unmarshal(data, &result)
	return result, nil
}

func orchestrate(task string, maxSteps int) string {
	messages := []Message{{Role: "system", Content: systemPrompt}, {Role: "user", Content: task}}
	for step := 0; step < maxSteps; step++ {
		resp, err := chat(messages)
		if err != nil {
			return "[error] " + err.Error()
		}
		content := resp["choices"].([]interface{})[0].(map[string]interface{})["message"].(map[string]interface{})["content"].(string)
		var parsed map[string]interface{}
		json.Unmarshal([]byte(content), &parsed)
		fmt.Printf("[step %d] %s\n", step+1, parsed["thought"])
		if final, ok := parsed["final_answer"]; ok {
			return final.(string)
		}
		stepPlan := parsed["step"].(map[string]interface{})
		workerName := stepPlan["worker"].(string)
		args := stepPlan["args"].(map[string]interface{})
		var result map[string]interface{}
		if fn, ok := workers[workerName]; ok {
			result = fn(args)
		} else {
			result = map[string]interface{}{"error": "unknown worker " + workerName}
		}
		messages = append(messages, Message{Role: "assistant", Content: content})
		resJSON, _ := json.Marshal(result)
		messages = append(messages, Message{Role: "user", Content: "[observation] " + string(resJSON)})
	}
	return "[timeout] 未能在限定步数内完成"
}

func main() {
	fmt.Println(orchestrate("查询北京天气，并将结果通过邮件发送给 bob@example.com", 5))
}

# mcp_server_stdio.py
import os
from pathlib import Path

from mcp.server.fastmcp import FastMCP

# pip install "mcp[cli]"
# SDK 负责协议协商、tools/list、tools/call 与 JSON Schema 生成；
# 业务代码只声明可用工具和自己的安全边界。
mcp = FastMCP("workspace-tools", json_response=True)

WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", ".")).resolve()

def resolve_inside_workspace(raw_path: str) -> Path:
    path = (WORKSPACE / raw_path).resolve()
    if path != WORKSPACE and WORKSPACE not in path.parents:
        raise ValueError("path escapes workspace")
    return path

@mcp.tool()
def read_file(path: str) -> dict:
    """Read a UTF-8 text file inside the configured workspace."""
    target = resolve_inside_workspace(path)
    return {"content": target.read_text(encoding="utf-8")}

@mcp.tool()
def list_files(directory: str = ".") -> dict:
    """List direct children inside the configured workspace."""
    target = resolve_inside_workspace(directory)
    return {"files": sorted(item.name for item in target.iterdir())}

if __name__ == "__main__":
    mcp.run(transport="stdio")

不建议在业务代码中手写旧版 initialize/JSON-RPC 分发器。使用官方 SDK，随后把路径边界、权限、审计、幂等与人审放在业务层；协议字段和版本协商由 SDK 跟进。

# mcp_client_stdio.py
import asyncio

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    params = StdioServerParameters(
        command="python3",
        args=["mcp_server_stdio.py"],
    )
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([tool.name for tool in tools.tools])
            result = await session.call_tool("list_files", {"directory": "."})
            print(result.content)

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

# minimal_vector_search.py
# 不依赖外部向量库，用纯 Python 实现 cosine similarity 检索
import math

def tokenize(text: str) -> set:
    return set(text.lower().split())

def vectorize(text: str, vocab: list) -> list:
    tokens = tokenize(text)
    return [1 if w in tokens else 0 for w in vocab]

def cosine(a: list, b: list) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(x * x for x in b))
    return dot / (norm_a * norm_b) if norm_a and norm_b else 0.0

class MinimalVectorStore:
    def __init__(self, docs: list):
        self.docs = docs
        self.vocab = sorted(set(w for d in docs for w in tokenize(d)))
        self.vectors = [vectorize(d, self.vocab) for d in docs]

    def search(self, query: str, top_k: int = 2) -> list:
        qv = vectorize(query, self.vocab)
        scored = [(cosine(qv, dv), doc) for dv, doc in zip(self.vectors, self.docs)]
        scored.sort(reverse=True)
        return scored[:top_k]

if __name__ == "__main__":
    docs = [
        "AI Agent 能调用工具完成复杂任务",
        "RAG 通过检索增强生成质量",
        "LLM 是大语言模型",
    ]
    store = MinimalVectorStore(docs)
    results = store.search("Agent 怎么调用工具", top_k=2)
    for score, doc in results:
        print(f"{score:.3f}  {doc}")

# agentic_rag.py
# Agent 根据问题决定检索次数和生成答案
import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", "sk-fake"))

DOCS = {
    "agent_basics": "AI Agent 能感知环境、调用工具、迭代决策。",
    "rag_intro": "RAG 通过向量检索把相关知识注入上下文。",
    "mcp_intro": "MCP 是 Agent 调用外部工具的标准协议。",
}

def retrieve(query: str) -> str:
    # 简化检索：关键词匹配
    results = []
    for k, v in DOCS.items():
        if any(w in v for w in query.lower().split()):
            results.append(f"[{k}] {v}")
    return "\n".join(results) if results else "未检索到相关内容"

SYSTEM_PROMPT = """你是 Agentic RAG。每一步输出 JSON：
{"thought": "...", "action": {"tool": "retrieve", "args": {"query": "..."}}}
若信息足够，输出：{"thought": "...", "final_answer": "..."}"""

def agentic_rag(question: str, max_steps: int = 3) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": question},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={"type": "json_object"},
        )
        content = resp.choices[0].message.content
        parsed = json.loads(content)
        print(f"[step {step + 1}] {parsed.get('thought', '')}")
        if "final_answer" in parsed:
            return parsed["final_answer"]
        action = parsed.get("action", {})
        if action.get("tool") == "retrieve":
            result = retrieve(action["args"]["query"])
            messages.append({"role": "assistant", "content": content})
            messages.append({"role": "user", "content": f"[observation] {result}"})
    return "[timeout]"

if __name__ == "__main__":
    print(agentic_rag("MCP 和 Agent 有什么关系？"))

# etclovg_config.py
# 用配置对象表达 ETCLOVG 七层约束
from dataclasses import dataclass, field

@dataclass
class ExecutionConfig:
    sandbox_type: str = "docker"
    timeout_seconds: int = 120
    max_memory_mb: int = 2048
    network_default: str = "deny"
    allow_domains: list = field(default_factory=lambda: ["api.github.com"])

@dataclass
class ToolingConfig:
    tools: list = field(default_factory=list)
    dynamic_visibility: bool = True
    schema_validation: bool = True

@dataclass
class ContextConfig:
    max_tokens: int = 8000
    system_prompt_budget: int = 1000
    tool_output_budget: int = 4000
    summary_interval: int = 5

@dataclass
class LifecycleConfig:
    max_steps: int = 10
    max_tool_calls: int = 20
    replan_enabled: bool = True

@dataclass
class ObservabilityConfig:
    trace_enabled: bool = True
    cost_tracking: bool = True
    log_level: str = "INFO"

@dataclass
class VerificationConfig:
    auto_test: bool = True
    failure_taxonomy: list = field(default_factory=lambda: [
        "planning", "tool_use", "context", "environment", "safety", "eval"
    ])

@dataclass
class GovernanceConfig:
    input_guard: bool = True
    action_guard: bool = True
    human_in_the_loop: bool = True
    audit_retention_days: int = 180

@dataclass
class HarnessConfig:
    execution: ExecutionConfig = field(default_factory=ExecutionConfig)
    tooling: ToolingConfig = field(default_factory=ToolingConfig)
    context: ContextConfig = field(default_factory=ContextConfig)
    lifecycle: LifecycleConfig = field(default_factory=LifecycleConfig)
    observability: ObservabilityConfig = field(default_factory=ObservabilityConfig)
    verification: VerificationConfig = field(default_factory=VerificationConfig)
    governance: GovernanceConfig = field(default_factory=GovernanceConfig)

if __name__ == "__main__":
    cfg = HarnessConfig()
    print(f"沙箱超时: {cfg.execution.timeout_seconds}s, 最大步数: {cfg.lifecycle.max_steps}")

# harness_loop.py
# 把 ETCLOVG 七层约束整合进 Agent 主循环
import time

class Harness:
    def __init__(self, config: dict):
        self.max_steps = config.get("max_steps", 10)
        self.timeout = config.get("timeout", 60)
        self.tool_whitelist = config.get("tools", [])
        self.start_time = time.time()

    def check_governance(self, action: dict) -> bool:
        if action["tool"] not in self.tool_whitelist:
            print(f"[G] 拦截未授权工具: {action['tool']}")
            return False
        return True

    def check_budget(self, step: int) -> bool:
        if step >= self.max_steps:
            print("[G] 超过最大步数")
            return False
        if time.time() - self.start_time > self.timeout:
            print("[G] 超过超时时间")
            return False
        return True

    def run(self, agent, task: str):
        step = 0
        while self.check_budget(step):
            step += 1
            print(f"\n[L] step {step}")
            action = agent.reason(task)
            print(f"[O] 计划动作: {action}")
            if not self.check_governance(action):
                break
            result = agent.execute(action)
            print(f"[V] 执行结果: {result}")
            if agent.is_done(result):
                return "success"
        return "failure"

class DummyAgent:
    def reason(self, task: str) -> dict:
        return {"tool": "read_file", "args": {"path": "note.txt"}}

    def execute(self, action: dict) -> str:
        return f"executed {action['tool']}"

    def is_done(self, result: str) -> bool:
        return "executed" in result

if __name__ == "__main__":
    harness = Harness({"max_steps": 3, "timeout": 10, "tools": ["read_file"]})
    print(harness.run(DummyAgent(), "读取文件"))

// HarnessConfig.java
public class HarnessConfig {
    public static class Execution {
        int timeoutSeconds = 120;
        int maxMemoryMb = 2048;
        String networkDefault = "deny";
    }

    public static class Lifecycle {
        int maxSteps = 10;
        int maxToolCalls = 20;
    }

    public static class Governance {
        boolean inputGuard = true;
        boolean actionGuard = true;
    }

    Execution execution = new Execution();
    Lifecycle lifecycle = new Lifecycle();
    Governance governance = new Governance();

    public boolean validate() {
        return execution.timeoutSeconds > 0
            && lifecycle.maxSteps > 0
            && lifecycle.maxToolCalls >= lifecycle.maxSteps;
    }

    public static void main(String[] args) {
        HarnessConfig cfg = new HarnessConfig();
        System.out.println("config valid: " + cfg.validate());
    }
}

# sandbox_config.py
# 声明式沙箱配置：最小权限 + 资源预算 + 网络白名单
import fnmatch

class Sandbox:
    def __init__(self, config: dict):
        self.fs_rules = config.get("filesystem", {})
        self.network = config.get("network", {})
        self.process = config.get("process", {})

    def allowed_file_action(self, path: str, action: str) -> bool:
        for deny in self.fs_rules.get("deny", []):
            if fnmatch.fnmatch(path, deny.get("path", "")) and action in deny.get("actions", []):
                return False
        for allow in self.fs_rules.get("allow", []):
            if fnmatch.fnmatch(path, allow.get("path", "")) and action in allow.get("actions", []):
                return True
        return self.fs_rules.get("default", "deny") == "allow"

    def allowed_domain(self, domain: str) -> bool:
        if self.network.get("default", "deny") == "allow":
            return True
        return domain in self.network.get("allowlist", [])

if __name__ == "__main__":
    cfg = {
        "filesystem": {
            "default": "deny",
            "allow": [{"path": "./src/**", "actions": ["read", "write"]}],
            "deny": [{"path": "./.env", "actions": ["read", "write"]}],
        },
        "network": {"default": "deny", "allowlist": ["api.github.com"]},
        "process": {"timeout_seconds": 120, "max_memory_mb": 2048},
    }
    sb = Sandbox(cfg)
    print(sb.allowed_file_action("./src/main.py", "read"))  # True
    print(sb.allowed_file_action("./.env", "read"))         # False
    print(sb.allowed_domain("api.github.com"))              # True
    print(sb.allowed_domain("evil.com"))                    # False

# governance_hooks.py
# 模拟 H1-H4 四个治理点
import re

class Governance:
    HIGH_RISK_TOOLS = {"delete_file", "send_email", "git_push"}

    def h1_input_guard(self, user_input: str) -> tuple:
        # 检查 prompt 注入
        if re.search(r"忽略.*指令|ignore previous", user_input, re.I):
            return False, "H1 拦截：疑似 prompt 注入"
        return True, "H1 通过"

    def h2_action_guard(self, tool: str, args: dict, risk_whitelist: set) -> tuple:
        if tool in self.HIGH_RISK_TOOLS and tool not in risk_whitelist:
            return False, f"H2 高风险：{tool} 需要审批"
        return True, "H2 通过"

    def h3_output_filter(self, output: str, secrets: list) -> str:
        for s in secrets:
            output = output.replace(s, "***")
        return output

    def h4_human_approval(self, tool: str, args: dict) -> bool:
        # 简化：模拟人审通过
        print(f"[H4] 请审批 {tool}({args})，模拟通过")
        return True

def run_with_governance(user_input: str, tool: str, args: dict, output: str, secrets: list):
    g = Governance()
    ok, msg = g.h1_input_guard(user_input)
    print(msg)
    if not ok:
        return
    ok, msg = g.h2_action_guard(tool, args, set())
    print(msg)
    if not ok:
        if not g.h4_human_approval(tool, args):
            return
    safe_output = g.h3_output_filter(output, secrets)
    print(f"执行结果（已脱敏）: {safe_output}")

if __name__ == "__main__":
    run_with_governance(
        "帮我发邮件", "send_email",
        {"to": "bob@example.com"},
        "邮件已发送，token: sk-abc123",
        ["sk-abc123"],
    )

# agent_evaluator.py
# 对 Agent episode 做结果评估、轨迹评估、失败分类
class AgentEvaluator:
    FAILURE_TYPES = ["planning", "tool_use", "context", "environment", "safety", "eval"]

    def evaluate_result(self, expected: str, actual: str) -> bool:
        return expected.lower().strip() == actual.lower().strip()

    def evaluate_trace(self, trace: list, max_steps: int) -> dict:
        steps = len(trace)
        loops = len(trace) - len(set(trace))
        return {
            "steps": steps,
            "exceeded_budget": steps > max_steps,
            "suspected_loops": loops,
        }

    def classify_failure(self, trace: list, error: str) -> str:
        if "tool" in error.lower():
            return "tool_use"
        if "timeout" in error.lower() or "memory" in error.lower():
            return "environment"
        if "permission" in error.lower() or "unauthorized" in error.lower():
            return "safety"
        return "planning"

if __name__ == "__main__":
    ev = AgentEvaluator()
    print(ev.evaluate_result("success", "success"))
    print(ev.evaluate_trace(["read", "think", "read", "think"], 5))
    print(ev.classify_failure([], "tool not found"))

# wiki_ingest.py
# 模拟 LLM Wiki 的 Ingest 流程：读来源 → 摘要 → 判断关联 → 建页
import json
import re
from datetime import datetime

class WikiIngest:
    def __init__(self, wiki: dict):
        self.wiki = wiki

    def parse_source(self, raw_text: str, source_path: str) -> dict:
        summary = raw_text[:200] + "..." if len(raw_text) > 200 else raw_text
        return {
            "source": source_path,
            "summary": summary,
            "entities": re.findall(r"`([^`]+)`", raw_text),
            "date": datetime.now().isoformat(),
        }

    def find_related(self, entities: list) -> list:
        related = []
        for title, page in self.wiki.items():
            if any(e in page.get("entities", []) for e in entities):
                related.append(title)
        return related

    def create_or_update(self, title: str, ingest: dict):
        if title in self.wiki:
            self.wiki[title]["sources"].append(ingest["source"])
            self.wiki[title]["updated"] = ingest["date"]
        else:
            self.wiki[title] = {
                "title": title,
                "summary": ingest["summary"],
                "entities": ingest["entities"],
                "sources": [ingest["source"]],
                "created": ingest["date"],
                "updated": ingest["date"],
            }

if __name__ == "__main__":
    wiki = {}
    ingest = WikiIngest(wiki)
    raw = "MCP 是 Model Context Protocol 的缩写，Agent 可以通过 MCP 调用外部工具。详见 `mcp-server-design`。"
    parsed = ingest.parse_source(raw, "raw/sessions/mcp-intro.md")
    related = ingest.find_related(parsed["entities"])
    print("相关页面:", related)
    ingest.create_or_update("MCP", parsed)
    print(json.dumps(wiki, ensure_ascii=False, indent=2))

# wiki_query.py
# 基于 wiki 页面生成带来源的回答
class WikiQuery:
    def __init__(self, wiki: dict):
        self.wiki = wiki

    def search(self, query: str) -> list:
        results = []
        for title, page in self.wiki.items():
            score = 0
            for word in query.lower().split():
                if word in title.lower() or word in page.get("summary", "").lower():
                    score += 1
            if score > 0:
                results.append((score, page))
        results.sort(reverse=True)
        return [p for _, p in results]

    def answer(self, query: str) -> str:
        pages = self.search(query)[:3]
        if not pages:
            return "未找到相关知识 (?)"
        lines = [f"关于 '{query}'："]
        for p in pages:
            lines.append(f"- {p['title']}: {p['summary']}")
            lines.append(f"  来源: {', '.join(p['sources'])}")
        return "\n".join(lines)

if __name__ == "__main__":
    wiki = {
        "MCP": {
            "title": "MCP",
            "summary": "MCP 是 Agent 调用外部工具的标准协议。",
            "sources": ["raw/sessions/mcp-intro.md"],
        },
        "Agent": {
            "title": "Agent",
            "summary": "Agent 是能感知环境、调用工具、迭代决策的 LLM 程序。",
            "sources": ["raw/sessions/agent-basics.md"],
        },
    }
    print(WikiQuery(wiki).answer("MCP 是什么"))

# wiki_lint.py
# 定期体检：孤儿页、缺来源页、过期页
from datetime import datetime, timedelta

class WikiLint:
    def __init__(self, wiki: dict, links: dict):
        self.wiki = wiki
        self.links = links  # title -> [linked_titles]

    def orphan_pages(self) -> list:
        referenced = set()
        for targets in self.links.values():
            referenced.update(targets)
        return [t for t in self.wiki if t not in referenced and t != "index"]

    def missing_sources(self) -> list:
        return [t for t, p in self.wiki.items() if not p.get("sources")]

    def stale_pages(self, days: int = 90) -> list:
        cutoff = datetime.now() - timedelta(days=days)
        stale = []
        for t, p in self.wiki.items():
            updated = p.get("updated", "")
            if updated:
                try:
                    if datetime.fromisoformat(updated) < cutoff:
                        stale.append(t)
                except ValueError:
                    pass
        return stale

if __name__ == "__main__":
    wiki = {
        "MCP": {"sources": ["s1"], "updated": "2026-07-01"},
        "Orphan": {"sources": ["s2"], "updated": "2026-07-01"},
        "NoSource": {"sources": [], "updated": "2026-07-01"},
        "Old": {"sources": ["s3"], "updated": "2026-01-01"},
    }
    links = {"index": ["MCP"], "MCP": []}
    lint = WikiLint(wiki, links)
    print("孤儿页:", lint.orphan_pages())
    print("缺来源:", lint.missing_sources())
    print("过期页:", lint.stale_pages())

from pydantic import BaseModel, Field
from typing import Literal
import hashlib

class QueryApprovalInput(BaseModel):
    ticket_id: str = Field(..., description="审批单号，形如 AP-20250710-001")
    fields: list[str] | None = Field(None, description="只返回指定字段，默认 summary")
    idempotency_key: str | None = Field(None, description="幂等键，写操作必填")

class ToolError(Exception):
    def __init__(self, code: str, message: str, retryable: bool, suggestion: str = ""):
        self.code = code
        self.message = message
        self.retryable = retryable
        self.suggestion = suggestion

async def query_approval_ticket(input: QueryApprovalInput):
    """
    查询审批单核心信息。只读工具，低风险。
    """
    if not input.ticket_id.startswith("AP-"):
        raise ToolError(
            code="INVALID_TICKET_ID",
            message=f"ticket_id 格式错误: {input.ticket_id}",
            retryable=False,
            suggestion="请确认单号以 AP- 开头",
        )
    # 实际查询数据库或 RPC
    data = {"ticket_id": input.ticket_id, "status": "pending", "owner": "carly"}
    if input.fields:
        data = {k: v for k, v in data.items() if k in input.fields}
    return {
        "ticket_id": input.ticket_id,
        "data": data,
        "cached": False,
        "cost_ms": 12,
    }

async def approve_ticket(input: QueryApprovalInput):
    """
    审批通过。写操作，需要幂等键 + Guardrails 授权。
    """
    if not input.idempotency_key:
        raise ToolError(
            code="MISSING_IDEMPOTENCY_KEY",
            message="写操作必须提供 idempotency_key",
            retryable=False,
        )
    # 幂等去重 + 权限判断
    # if already_approved(input.idempotency_key): return {"already_applied": True}
    # if not guardrails.allow("approve_ticket", input): raise ToolError(...)
    return {
        "ticket_id": input.ticket_id,
        "action": "approved",
        "idempotency_key": input.idempotency_key,
    }

from dataclasses import dataclass, field
from typing import Callable, Literal
import json

# ---------- 数据模型 ----------
@dataclass
class Tool:
    name: str
    description: str
    params_schema: dict
    fn: Callable[[dict], dict]

@dataclass
class AgentState:
    question: str
    messages: list[dict] = field(default_factory=list)
    turns: int = 0
    max_turns: int = 5
    done: bool = False

# ---------- 模拟 LLM 调用 ----------
def call_llm(messages: list[dict]) -> dict:
    """
    真实环境应替换为 OpenAI/Claude/Moonshot 等 SDK。
    这里用一个简单规则模拟：如果问题包含数字就调用 calc，否则直接回答。
    """
    last_user = next((m["content"] for m in reversed(messages) if m["role"] == "user"), "")
    if "35" in last_user or "12" in last_user:
        return {
            "type": "tool_call",
            "tool": "calc",
            "arguments": {"expr": "35 + 12"},
        }
    return {"type": "answer", "content": "这个问题不需要工具，直接回答即可。"}

# ---------- 工具实现 ----------
def calc_tool(args: dict) -> dict:
    expr = args.get("expr", "")
    try:
        result = eval(expr, {"__builtins__": {}}, {})
        return {"ok": True, "result": result, "expr": expr}
    except Exception as e:
        return {"ok": False, "error": str(e), "expr": expr}

TOOLS = {
    "calc": Tool(
        name="calc",
        description="计算数学表达式，例如 '35 + 12'。",
        params_schema={"type": "object", "properties": {"expr": {"type": "string"}}},
        fn=calc_tool,
    )
}

# ---------- Agent 运行时 ----------
def plan(state: AgentState) -> AgentState:
    state.messages.append({"role": "system", "content": "你是一个会调用工具的助手。每步输出 tool_call 或 answer。"})
    state.messages.append({"role": "user", "content": state.question})
    return state

def act(state: AgentState) -> AgentState:
    response = call_llm(state.messages)
    state.messages.append({"role": "assistant", "content": json.dumps(response, ensure_ascii=False)})
    state.turns += 1

    if response["type"] == "answer":
        state.done = True
        return state

    tool_name = response.get("tool")
    args = response.get("arguments", {})
    if tool_name not in TOOLS:
        state.messages.append({
            "role": "tool",
            "content": json.dumps({"error": f"未知工具: {tool_name}"}, ensure_ascii=False),
        })
        return state

    result = TOOLS[tool_name].fn(args)
    state.messages.append({
        "role": "tool",
        "content": json.dumps(result, ensure_ascii=False),
    })
    return state

def reflect(state: AgentState) -> AgentState:
    """
    反射：如果上一轮工具调用失败，让 LLM 自己反思并尝试修正。
    """
    last_tool_msg = next((m for m in reversed(state.messages) if m["role"] == "tool"), None)
    if last_tool_msg is None:
        return state
    payload = json.loads(last_tool_msg["content"])
    if payload.get("ok") is False:
        state.messages.append({
            "role": "system",
            "content": "上一轮工具调用失败，请反思原因并修正参数后重试。",
        })
    return state

def should_continue(state: AgentState) -> Literal["act", "finish"]:
    if state.done or state.turns >= state.max_turns:
        return "finish"
    return "act"

def run_agent(question: str) -> AgentState:
    state = AgentState(question=question)
    state = plan(state)
    while True:
        state = act(state)
        if should_continue(state) == "finish":
            break
        state = reflect(state)
    return state

if __name__ == "__main__":
    final_state = run_agent("35 加 12 等于多少？")
    print(f"turns={final_state.turns}, done={final_state.done}")
    print(json.dumps(final_state.messages, ensure_ascii=False, indent=2))