while not reached_max_steps:
    decision = llm.generate(history + system_prompt, tools=tool_schemas)
    if decision.wants_tool:
        result = execute_tool(decision.tool_name, decision.tool_args)
        history.append(result)
    else:
        return decision.final_text

import inspect
from typing import Any, Callable

class ToolRegistry:
    """轻量级工具注册中枢：负责 Schema 管理、动态分发与安全防护"""

    def __init__(self):
        self._schemas: dict[str, dict] = {}
        self._handlers: dict[str, Callable] = {}

    def register(self, name: str, description: str, parameters: dict, func: Callable):
        """显式注册工具"""
        self._schemas[name] = {
            "name": name,
            "description": description,
            "input_schema": parameters,
        }
        self._handlers[name] = func

    def tool(self, name: str, description: str, parameters: dict):
        """装饰器注册语法糖"""
        def decorator(func: Callable):
            self.register(name, description, parameters, func)
            return func
        return decorator

    def get_schemas(self) -> list[dict]:
        """导出所有提供给大模型的工具 JSON Schema 列表"""
        return list(self._schemas.values())

    def call(self, name: str, args: dict) -> Any:
        """根据名称动态执行目标本地函数，带有完善的异常兜底"""
        if name not in self._handlers:
            return f"执行失败：未找到名为 '{name}' 的已注册工具"
        try:
            handler = self._handlers[name]
            return handler(**args)
        except Exception as e:
            return f"工具 '{name}' 运行时抛出异常: {e}"

# 全局工具注册实例
registry = ToolRegistry()

@registry.tool(
    name="get_weather",
    description="查询指定城市的实时气象数据",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名称，例如北京、上海"},
        },
        "required": ["city"],
    },
)
def get_weather(city: str) -> str:
    mock_db = {
        "北京": "晴空万里，23°C，北风二级",
        "上海": "阴有阵雨，19°C，空气湿度 85%",
        "深圳": "多云，27°C，体感舒适",
    }
    return mock_db.get(city, f"{city} 暂无气象台站数据")


@registry.tool(
    name="calculate",
    description="高精度数学表达式求值计算器",
    parameters={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "标准数学运算表达式，例如 '12 * 8 + 4'"},
        },
        "required": ["expression"],
    },
)
def calculate(expression: str) -> str:
    # 防范代码注入：仅允许纯数学字符
    allowed_chars = set("0123456789+-*/()., ")
    if not all(c in allowed_chars for c in expression):
        return "安全校验失败：表达式包含不合法字符"
    try:
        # 在隔离的作用域内计算
        res = eval(expression, {"__builtins__": None}, {})
        return str(res)
    except Exception as e:
        return f"数学运算错误: {e}"


@registry.tool(
    name="query_knowledge",
    description="在本地系统技术文档库中检索指定名词或规范",
    parameters={
        "type": "object",
        "properties": {
            "keyword": {"type": "string", "description": "需要查询的关键词"},
        },
        "required": ["keyword"],
    },
)
def query_knowledge(keyword: str) -> str:
    docs = {
        "agentscope": "AgentScope 是阿里巴巴达摩院开源的多智能体框架，主打以消息（Msg）为一等公民和 RPC 分布式部署。",
        "semantic kernel": "Semantic Kernel 是微软主导的企业级 AI SDK，深度集成 C#/.NET 与依赖注入插件体系。",
        "eino": "Eino 是字节跳动开源的 Go 语言 AI 框架，针对生产级高并发与全链路流式设计。",
    }
    kw_lower = keyword.lower()
    for k, v in docs.items():
        if k in kw_lower:
            return v
    return f"文档库中未收录与 '{keyword}' 匹配的知识条目"

import os
from anthropic import Anthropic

class AgentHarness:
    """轻量级 Agent 运行底座"""

    def __init__(
        self,
        system_prompt: str,
        model: str = "claude-opus-4-5",
        max_iterations: int = 8,
        verbose: bool = True,
    ):
        self.client = Anthropic()
        self.system_prompt = system_prompt
        self.model = model
        self.max_iterations = max_iterations
        self.verbose = verbose
        self.history: list[dict] = []

    def _log(self, text: str):
        if self.verbose:
            print(text)

    def run(self, user_prompt: str) -> str:
        """执行单次任务直至生成最终结论或触发步数熔断"""
        # 将用户目标追加至历史对话列表
        self.history.append({"role": "user", "content": user_prompt})

        for step in range(1, self.max_iterations + 1):
            self._log(f"\n[🔄 步数 {step}/{self.max_iterations}] 正在等待大模型规划决策...")

            response = self.client.messages.create(
                model=self.model,
                max_tokens=2048,
                system=self.system_prompt,
                tools=registry.get_schemas(),
                messages=self.history,
            )

            # 分支 1：模型判断需要调用外部工具
            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        self._log(f"  👉 [调用工具] {block.name}(参数: {block.input})")
                        exec_res = registry.call(block.name, block.input)
                        self._log(f"  👈 [工具输出] {exec_res}")

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(exec_res),
                        })

                # 将模型的工具调用意图与执行结果作为多轮历史写回
                self.history.append({"role": "assistant", "content": response.content})
                self.history.append({"role": "user", "content": tool_results})

            # 分支 2：模型认为已获得足够信息，输出最终文本回答
            elif response.stop_reason == "end_turn":
                final_answer = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        final_answer += block.text

                self.history.append({"role": "assistant", "content": final_answer})
                self._log("  ✅ [任务完成] 模型已生成最终答复。")
                return final_answer

        # 防御兜底：防止提示词死循环导致无限消耗 API Token
        warning = "⚠️ [熔断触发] 已达到最大迭代步数限制，当前任务被强制中断。"
        self._log(warning)
        return warning

    def reset(self):
        """清空会话历史"""
        self.history = []

if __name__ == "__main__":
    harness = AgentHarness(
        system_prompt="""你是一名严谨的智能技术助手。
你可以自主调用天气查询、数学计算与技术知识库检索工具来协助回答问题。
必须基于工具返回的事实给出答案，严禁凭空捏造。""",
        verbose=True,
    )

    query = (
        "帮我查一下北京现在的天气；"
        "另外算一下 128 乘以 3.5 加上 42 等于多少；"
        "最后在文档库里检索一下 AgentScope 是什么。"
    )

    print("=" * 60)
    print("用户指令:", query)
    print("=" * 60)

    final_result = harness.run(query)

    print("\n" + "=" * 60)
    print("智能体最终输出：\n", final_result)
    print("=" * 60)

import concurrent.futures

def execute_tools_parallel(tool_calls: list) -> list[dict]:
    """多线程并发执行工具列表"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_map = {
            executor.submit(registry.call, tc.name, tc.input): tc
            for tc in tool_calls
        }
        for future in concurrent.futures.as_completed(future_map):
            tc = future_map[future]
            try:
                res = future.result()
            except Exception as e:
                res = f"线程池执行失败: {e}"
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": str(res),
            })
    return results