1
class AgentLoop:
2
    """
3
    Agent的感知-规划-行动闭环的抽象表示
4
    """
5
    def __init__(self, llm, memory, tools, planner):
6
        self.llm = llm          # 推理引擎
7
        self.memory = memory    # 记忆系统
8
        self.tools = tools      # 工具集
9
        self.planner = planner  # 规划模块
10

11
    def run(self, user_goal: str) -> str:
12
        """执行Agent闭环"""
13
        # 1. 感知：接收用户目标
14
        current_state = {"goal": user_goal, "history": []}
15

16
        while not self._is_goal_achieved(current_state):
17
            # 2. 规划：生成行动计划
18
            plan = self.planner.plan(current_state)
19

20
            # 3. 行动：执行计划中的每一步
21
            for action in plan:
22
                result = self._execute_action(action)
23
                # 4. 观察与反馈：更新状态
24
                current_state["history"].append({
25
                    "action": action,
26
                    "result": result
27
                })
28
                self.memory.update(current_state)
29

30
        return self._generate_response(current_state)

1
class LLMReasoningEngine:
2
    """
3
    LLM作为中央推理引擎的抽象
4
    """
5
    def __init__(self, model, system_prompt: str):
6
        self.model = model
7
        self.system_prompt = system_prompt
8

9
    def reason(self,
10
               user_query: str,
11
               memory_context: str,
12
               tool_descriptions: str) -> dict:
13
        """
14
        基于当前上下文进行推理，决定下一步行动
15

16
        Returns:
17
            {
18
                "thought": "推理过程",
19
                "action": "要执行的动作",
20
                "action_input": "动作的参数"
21
            }
22
        """
23
        prompt = self._build_prompt(
24
            system=self.system_prompt,
25
            memory=memory_context,
26
            tools=tool_descriptions,
27
            query=user_query
28
        )
29

30
        response = self.model.generate(prompt)
31
        return self._parse_decision(response)
32

33
    def _build_prompt(self, system, memory, tools, query):
34
        """构建包含系统指令、记忆、工具描述和用户查询的完整提示"""
35
        return f"""
36
        {system}
37

38
        记忆:
39
        {memory}
40

41
        可用工具:
42
        {tools}
43

44
        用户请求: {query}
45

46
        请逐步推理并决定下一步行动。
47
        """

1
from typing import List, Dict, Any
2
from dataclasses import dataclass
3

4
@dataclass
5
class SubTask:
6
    """子任务的数据结构"""
7
    id: str
8
    description: str
9
    required_tools: List[str]
10
    dependencies: List[str]  # 依赖的其他子任务ID
11
    status: str  # "pending", "running", "completed", "failed"
12

13
class Planner:
14
    """
15
    Agent的规划模块
16
    """
17
    def __init__(self, llm):
18
        self.llm = llm
19

20
    def plan(self, goal: str, available_tools: List[str]) -> List[SubTask]:
21
        """
22
        将高层目标分解为子任务序列
23

24
        数学上，这等价于求解：
25
        plan = argmax_{p} P(p | goal, tools)
26
        其中p是任务分解方案
27
        """
28
        # 1. 让LLM进行任务分解
29
        decomposition = self._decompose_goal(goal, available_tools)
30

31
        # 2. 构建子任务依赖图
32
        tasks = self._build_dependency_graph(decomposition)
33

34
        # 3. 拓扑排序确定执行顺序
35
        ordered_tasks = self._topological_sort(tasks)
36

37
        return ordered_tasks
38

39
    def _decompose_goal(self, goal: str, tools: List[str]) -> List[Dict]:
40
        """调用LLM进行任务分解"""
41
        prompt = f"""
42
        目标: {goal}
43
        可用工具: {tools}
44

45
        请将上述目标分解为3-5个可执行的子任务。
46
        每个子任务需要说明: 任务描述、所需工具、依赖的前置任务。
47
        """
48
        response = self.llm.generate(prompt)
49
        return self._parse_decomposition(response)
50

51
    def _topological_sort(self, tasks: List[SubTask]) -> List[SubTask]:
52
        """
53
        基于依赖关系进行拓扑排序
54

55
        这确保了: 如果任务B依赖任务A，则A在B之前执行
56
        """
57
        # 实现标准的拓扑排序算法
58
        # 时间复杂度: O(V + E)，其中V是任务数，E是依赖边数
59
        ...

1
{
2
    "name": "search",
3
    "arguments": {
4
        "query": "2024年诺贝尔物理学奖得主"
5
    }
6
}

1
from typing import Any, Dict, Callable
2
from dataclasses import dataclass
3
import json
4

5
@dataclass
6
class Tool:
7
    """
8
    工具的标准化抽象接口
9
    """
10
    name: str                          # 工具名称
11
    description: str                   # 功能描述
12
    parameters: Dict[str, Any]         # 参数的JSON Schema
13
    func: Callable                     # 实际执行的函数
14

15
    def to_openai_schema(self) -> Dict:
16
        """转换为OpenAI Function Calling格式"""
17
        return {
18
            "type": "function",
19
            "function": {
20
                "name": self.name,
21
                "description": self.description,
22
                "parameters": self.parameters
23
            }
24
        }
25

26
    def execute(self, **kwargs) -> Any:
27
        """执行工具调用"""
28
        return self.func(**kwargs)
29

30

31
class ToolRegistry:
32
    """
33
    工具注册中心：管理所有可用工具
34
    """
35
    def __init__(self):
36
        self._tools: Dict[str, Tool] = {}
37

38
    def register(self, tool: Tool):
39
        """注册一个工具"""
40
        self._tools[tool.name] = tool
41

42
    def get_schemas(self) -> List[Dict]:
43
        """获取所有工具的OpenAI schema"""
44
        return [tool.to_openai_schema() for tool in self._tools.values()]
45

46
    def execute(self, name: str, arguments: Dict) -> Any:
47
        """根据名称和参数执行工具"""
48
        if name not in self._tools:
49
            raise ValueError(f"Unknown tool: {name}")
50
        return self._tools[name].execute(**arguments)
51

52

53
# 使用示例：注册一个搜索工具
54
def search_function(query: str) -> str:
55
    """模拟搜索功能"""
56
    return f"搜索结果: 关于'{query}'的信息..."
57

58
search_tool = Tool(
59
    name="search",
60
    description="搜索互联网获取信息",
61
    parameters={
62
        "type": "object",
63
        "properties": {
64
            "query": {
65
                "type": "string",
66
                "description": "搜索关键词"
67
            }
68
        },
69
        "required": ["query"]
70
    },
71
    func=search_function
72
)
73

74
registry = ToolRegistry()
75
registry.register(search_tool)