LangChain与AutoGPT启发下：从零手写Agent循环控制与记忆管理

import json
from typing import Dict, Any, List, Optional

class AgentCore:
    def __init__(self, llm_client, memory_manager, tool_registry, max_iterations=5):
        self.llm = llm_client
        self.memory = memory_manager
        self.tools = tool_registry
        self.max_iterations = max_iterations

    def run(self, user_query: str) -> Dict[str, Any]:
        """主循环：思考 -> 行动 -> 观察，直到满足终止条件"""
        self.memory.add_user_message(user_query)
        iteration = 0

        while iteration  List[Dict[str, str]]:
        """返回完整的上下文，system prompt永远在最前"""
        return [{"role": "system", "content": self.system_prompt}] + self.messages

    def _trim_to_budget(self):
        """核心：从最旧的消息开始删，直到总token数低于预算"""
        while self._count_tokens() > self.token_budget and len(self.messages) > 2:
            # 删除最旧的一条消息（但保留最后两条用户消息作为基础上下文）
            self.messages.pop(0)

    def _count_tokens(self) -> int:
        total = len(self.encoder.encode(self.system_prompt))
        for msg in self.messages:
            total += len(self.encoder.encode(msg["content"]))
            if msg["role"] == "tool":
                # tool消息内容可能超长，需要额外附加的token
                total += 20  # 经验值：函数调用ID和结构开销
        return total

import inspect
import functools
from typing import get_type_hint

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Dict] = {}
        self._functions: Dict[str, callable] = {}

    def register(self, name: str, description: str):
        """装饰器：自动从函数签名生成OpenAI tool schema"""
        def decorator(func):
            # 获取类型标注，生成parameters schema
            hints = get_type_hints(func)
            signature = inspect.signature(func)
            properties = {}
            required = []

            for param_name, param in signature.parameters.items():
                if param_name == 'return':
                    continue
                if hasattr(param, 'default') and param.default is not inspect.Parameter.empty:
                    # 有默认值则为可选参数
                    properties[param_name] = {"type": self._map_type(hints.get(param_name, str)), "description": f"Parameter {param_name}"}
                else:
                    required.append(param_name)
                    properties[param_name] = {"type": self._map_type(hints.get(param_name, str)), "description": f"Parameter {param_name}"}

            schema = {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": {
                        "type": "object",
                        "properties": properties,
                        "required": required
                    }
                }
            }

            self._tools[name] = schema
            self._functions[name] = func

            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper
        return decorator

    def get_function_schemas(self) -> List[Dict]:
        return [t["function"] for t in self._tools.values()]  # 注意：OpenAI API需要的是function列表

    def execute(self, name: str, **kwargs) -> Any:
        if name not in self._functions:
            raise ValueError(f"Tool {name} not registered")
        return self._functions[name](**kwargs)

    @staticmethod
    def _map_type(python_type) -> str:
        mapping = {int: "integer", float: "number", str: "string", bool: "boolean"}
        return mapping.get(python_type, "string")