1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

D:\agent_project\
├── agent\                # Agent 层：调度大脑
│   ├── __init__.py       # 暴露 fc_loop / auto_select 等入口
│   ├── main.py           # 组装所有层 + 交互循环
│   ├── core.py           # Function Calling 主循环 fc_loop
│   ├── router.py         # 选模型 + 选推理模式
│   └── modes\            # 四种推理模式
│       ├── cot.py        # 思维链
│       ├── few_shot.py   # 少样本
│       ├── react.py      # 工具调用
│       └── tot.py        # 树状思维
├── llm\                  # LLM 层：模型抽象与封装
│   ├── base.py           # 抽象接口 + 配置读取
│   ├── real_llm.py       # 非流式客户端（带 .env + summarize）
│   └── stream_llm.py     # 流式客户端
├── memory\               # 记忆层：短期 + 长期 + 压缩
│   ├── paths.py          # 落盘路径集中管理
│   ├── shortmemory.py    # 短期滑动窗口
│   ├── long_term.py      # 长期向量库（Chroma）
│   ├── compress.py       # 滚动摘要压缩
│   └── memory_manager.py # 统一调度者
├── tools\                # 工具层：工具定义 + 注册 + 执行
│   ├── base.py           # 工具基类
│   ├── registry.py       # 动态注册表
│   ├── executor.py       # 带"自愈"的执行器
│   ├── weather.py        # 查天气
│   ├── calculator.py     # 计算器
│   ├── save_preference.py# 存长期偏好
│   └── recall_memory.py  # 召回长期记忆
├── config.json           # providers + tools 清单
└── .env                  # 密钥（不进 git）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

                        用户输入 "你: ..."
                                 │
                                 ▼
        ┌─────────────────────────────────────────────┐
        │              agent/main.py  (组装口)          │
        │  1. mm.add(user_msg, persist=是否长期事实)      │
        │  2. ctx = mm.get_context(query=用户问题)        │
        │  3. out = fc_loop(llm, ctx, registry, executor)│
        │  4. mm.add(assistant_msg)                      │
        │  5. mm.maybe_compress()  ← 超阈值才压缩         │
        └───────────────────────┬─────────────────────┘
                                │
            ┌───────────────────┼───────────────────┐
            ▼                   ▼                   ▼
      ┌──────────┐       ┌──────────────┐    ┌─────────────┐
      │ 记忆层    │       │  Agent 层     │    │  工具层      │
      │ Memory   │◄──────┤ fc_loop 循环  │───►│ 调 get_weather│
      │ Manager  │       │ router 选模式 │    │ 调 calculator│
      └────┬─────┘       └──────┬───────┘    └──────┬──────┘
           │                    │                   │
           ▼                    ▼                   ▼
     短期窗口 + 长期向量库    RealLLM / StreamLLM    依赖注入 mm
           │                    (LLM 层)             (MemoryManager)
           └──────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# llm/base.py —— LLM 抽象接口
from pathlib import Path
import json

# 配置路径：从"llm/ 的上级目录"找 config.json
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"

def load_provider_config(provider: str) -> dict:
    with open(CONFIG_PATH, encoding="utf-8") as f:
        return json.load(f)["providers"][provider]

class LLMClient:
    def chat(self, messages, tools=None):
        raise NotImplementedError("子类必须实现 chat()")
    def chat_stream(self, messages, **kwargs):
        raise NotImplementedError("子类必须实现 chat_stream()")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# llm/real_llm.py —— 非流式客户端
import os
from pathlib import Path
from openai import OpenAI
from dotenv import load_dotenv
from .base import LLMClient, load_provider_config

# 加载项目根目录的 .env（让 DEEPSEEK_API_KEY 等从文件读取，优先级高于系统环境变量）
load_dotenv(Path(__file__).resolve().parent.parent / ".env", override=True)

class RealLLM(LLMClient):
    def __init__(self, provider="deepseek"):
        cfg = load_provider_config(provider)          # 从 config.json 读
        self.client = OpenAI(api_key=os.getenv(cfg["key_env"]),
                             base_url=cfg["base_url"])
        self.model = cfg["model"]
        print(f"实际使用的 provider={provider}, "
              f"key 已加载={'是' if os.getenv(cfg['key_env']) else '否'}")

    def chat(self, messages, tools=None,**kwargs) -> str:
        """普通对话：返回纯文本（demo / 对话循环用）"""
        resp = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=tools,
            max_tokens=kwargs.get("max_tokens"))
        return resp.choices[0].message.content

    def chat_for_tools(self, messages, tools):
        """工具对话：返回完整 message 对象（fc_loop 用，能拿到 tool_calls）"""
        resp = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=tools)
        return resp.choices[0].message

    def summarize(self, text: str) -> str:
        """把一段对话文本压缩成简短摘要（记忆压缩模块 compress 调用）"""
        messages = [
            {"role": "system", "content": "你是一个摘要助手。请把下面的对话浓缩成简洁的中文要点，"
                                           "保留关键事实、用户偏好与决策，不要添加新内容。"},
            {"role": "user", "content": text},
        ]
        return self.chat(messages, max_tokens=500)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# llm/stream_llm.py —— 流式客户端
import os
from openai import OpenAI
from .base import LLMClient, load_provider_config

class StreamLLM(LLMClient):
    def __init__(self, provider="deepseek"):
        cfg = load_provider_config(provider)
        self.client = OpenAI(api_key=os.getenv(cfg["key_env"]),
                             base_url=cfg["base_url"])
        self.model = cfg["model"]

    def chat_stream(self, messages, **kwargs):
        """流式：yield 逐段吐（边收边显示）"""
        stream = self.client.chat.completions.create(
            model=self.model, messages=messages, stream=True,
            reasoning_effort=kwargs.get("reasoning_effort", "low"),
            extra_body={"thinking": {"type": "enabled"}},
            max_tokens=kwargs.get("max_tokens", 500),
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:                     # 过滤空 chunk（思考过程不吐）
                yield delta

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

{
  "providers": {
    "deepseek": {
      "base_url": "https://api.deepseek.com",
      "key_env": "DEEPSEEK_API_KEY",
      "model": "deepseek-v4-flash"
    },
    "kimi": {
      "base_url": "https://api.moonshot.cn/v1",
      "key_env": "MOONSHOT_API_KEY",
      "model": "kimi-k2.6"
    }
  },
  "tools": [
    { "module": "tools.weather", "class": "WeatherTool" },
    { "module": "tools.calculator", "class": "CalculatorTool" },
    { "module": "tools.save_preference", "class": "SavePreferenceTool" },
    { "module": "tools.recall_memory", "class": "RecallMemoryTool" }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

"""集中管理所有落盘路径, 避免各处硬编码 './xxx' (相对路径会随 cwd 漂移)."""
# ============ 隐私分流（设计约定, 不是代码自动给的） ============
# 🔒 永远留本地、绝不进 prompt: 用户真实姓名 / 联系方式 / 密钥 / 明文敏感偏好
# 🌐 会发到云端 LLM API: 你主动塞进 system/user 消息的"召回结果"
# ⚠️ 敏感事实: 存本地, 但召回时不自动进 prompt, 需用户显式确认才带上
# ==============================================================

import os
BASE = os.path.dirname(os.path.abspath(__file__))
MEM_STORE   = os.path.join(BASE, "mem_store")    # Chroma 向量库
PROFILE_DB  = os.path.join(BASE, "memory.db")    # SQLite 用户画像
FACTS_JSON  = os.path.join(BASE, "facts.json")   # 纯本地简单事实

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class ShortTermMemory:
    def __init__(self,window=10,max_tokens:int=500,system_roles=("system",)):
        self.window=window          # 聊天窗口大小
        self.max_tokens=max_tokens  # 最大 tokens
        self.system_roles=system_roles
        self.buffer=[]              # 消息缓存

    def _tokens(self,msgs) ->int:   # 估算 token 数
        return sum(max(1,len(m.get("content",""))//3)for m in msgs)

    def add(self,msg:dict):         # 消息缓存
        self.buffer.append(msg)

    def context(self)->list[dict]:  # 返回最近 n 条消息，滑动窗口
        sys_msgs=[m for m in self.buffer if m.get("role") in self.system_roles]
        others=[m for m in self.buffer if m.get("role") not in self.system_roles]
        return sys_msgs+others[-self.window:]

    def trim(self):                 # 消息裁剪
        sys_msgs= [m for m in self.buffer if m.get("role") in self.system_roles]
        others =[m for m in self.buffer if m.get("role") not in self.system_roles]
        while self._tokens(others) >self.max_tokens and len(others)>1:
            others.pop(0)
        self.buffer=sys_msgs+others
        return self.buffer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

'''
长期记忆 = 关电脑也不丢
长期记忆 = 单文档 RAG + 落盘
    嵌入/相似度召回  → 只是把"文档"换成"用户事实"且持久化
pip install chromadb
'''
import chromadb
from pathlib import Path
import chromadb, time
from memory.paths import MEM_STORE

class LongTermMemory:
    def __init__(self, path=MEM_STORE, collection_name="long_term"):
        self.client = chromadb.PersistentClient(path=path)  # 绝对路径
        self.col = self.client.get_or_create_collection(collection_name)

    def add_fact(self, text: str, fid: str, private: bool = False):
        """存一条长期事实。private=True 表示敏感, 召回时默认不出网。"""
        self.col.add(ids=[fid], documents=[text],
                     metadatas=[{"private": private, "ts": time.time()}])

    def recall(self,query:str,k:int=3,include_private=False)->list[str]:
        """召回相关事实。默认跳过 private=True 的敏感事实(不进 prompt)。"""
        res=self.col.query(query_texts=[query],n_results=k)
        facts=[]
        for doc,meta in zip(res["documents"][0],res["metadatas"][0]):
            if  meta and meta.get("private") and not include_private:
                continue
            facts.append(doc)
        return facts

    def delete(self,fid:str):
        self.col.delete(ids=[fid])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

"""
上下文压缩——>一条滚动摘要，永远只有一条
"""

def _estimate_tokens(text:str) ->int:#token估算
    """
    估算tokens，达到上限压缩一次。
    中文为主：1 字≈1 token，直接用字符数估算（原 *0.6 严重低估中文，
    导致上下文要累积到 2500+ 字才触发压缩，实际等于不压缩）。
    """
    return max(1, len(text))

def _fmt(messages) ->str:#历史消息拼接
    """消息列表，摘要原材料"""
    return "\n".join(f"{m['role']}:{m['content']}" for m in messages)

def compress(messages:list[dict],llm,keep:int=4) ->list[dict]:#压缩逻辑函数
    """滚动压缩版本"""
    sys_msgs=[m for m in messages if m.get("role")=="system" and not m.get("is_summary")]
    old_sums=[m for m in messages if m.get("is_summary")]
    others=[m for m in messages if m.get("role")!="system" and not m.get("is_summary")]
    if len((others))<=keep:
        return messages
    old,recent=others[:-keep],others[-keep:]

    # 滚动关键：旧摘要 + 新滚出窗口的原文 → 合并成一条新摘要
    text=_fmt(old)
    if old_sums:
        text=f"已有摘要:{old_sums[-1]['content']}\n新增对话：\n{text}"
    summary=llm.summarize(text)
    summary_msg={"role":"system","content":f"历史摘要：{summary}","is_summary":True}
    return sys_msgs + [summary_msg] + recent

def maybe_compress(messages,llm,max_tokens:int=3000,keep=4):#自动压缩检测
    """自动触发，超阈值才压缩，没超原样返回"""
    if (_estimate_tokens(_fmt(messages))>max_tokens and len(messages)> keep+1):
        return compress(messages,llm,keep=keep)
    return messages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

"""
记忆调度：把三个记忆模块拼接起来，对外只暴露 add / get_context / maybe_compress / recall
- get_context 里召回结果是临时拼接、不写回 buffer → 避免"每次召回都追加 system 导致堆积"
- maybe_compress 用 `msgs is not self.stm.buffer` 判断"真的压了才写回" → 没超阈值不碰 buffer
- add_fact 的 fid 用内容 md5 哈希 → 稳定、不重复、跨进程
"""
from .compress import compress, maybe_compress as _auto_compress
from .shortmemory import ShortTermMemory
from .long_term import LongTermMemory
import hashlib

def _fid(text: str) -> str:
    """内容哈希生成稳定ID：不重复、可跨进程（比 hash() 稳，hash 有随机盐）"""
    return "f" + hashlib.md5(text.encode("utf-8")).hexdigest()[:12]

class MemoryManager:
    def __init__(self, window=8, max_tokens=2000, keep=4, llm=None):
        self.stm = ShortTermMemory(window=window, max_tokens=max_tokens)
        self.ltm = LongTermMemory()
        self.keep, self.max_tokens, self.llm = keep, max_tokens, llm

    def add(self, msg: dict, persist: bool = False):
        self.stm.add(msg)
        # 只有明确是用户事实才落长期（persist 默认 False，省 embedding 开销）
        if persist and msg.get("role") == "user":
            self.ltm.add_fact(msg["content"], fid=_fid(msg["content"]))

    def get_context(self, query: str = None) -> list[dict]:
        base = self.stm.context()  # [系统提示] + [窗口消息]
        if not query:
            return base
        facts = self.ltm.recall(query, k=3)
        if not facts:
            return base
        facts_msg = {"role": "system", "content": "已知用户长期事实：" + "|".join(facts)}
        sys_msgs = [m for m in base if m.get("role") in self.stm.system_roles]
        others = [m for m in base if m.get("role") not in self.stm.system_roles]
        return sys_msgs + [facts_msg] + others  # 系统提示 + 长期事实 + 窗口

    def maybe_compress(self, llm=None):
        llm = llm or self.llm
        if llm is None:
            return self.stm.buffer
        msgs = _auto_compress(self.stm.buffer, llm,
                              max_tokens=self.max_tokens, keep=self.keep)
        if msgs is not self.stm.buffer:  # 真压缩了才写回
            self.stm.buffer = msgs
        return self.stm.buffer

    def recall(self, query: str, k: int = 3) -> list[str]:
        return self.ltm.recall(query, k=k)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class BaseTool:
    name: str = ""              # 工具名（模型调用时用）
    description: str = ""       # 给模型看的说明（决定它会不会调）
    parameters: dict = {}       # JSON Schema（声明参数格式）

    def __init__(self, **kwargs):
        pass   # 允许注册时传入 mm=... 等依赖，不需要的工具直接忽略

    def execute(self, **kwargs) -> str:
        raise NotImplementedError("子类必须实现 execute()")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# 读 config.json 动态注册工具
import json, importlib
from .base import BaseTool

class ToolRegistry:
    def __init__(self, config_path: str, deps: dict = None):
        self._deps = deps or {}          # 依赖注入：{"mm": MemoryManager实例}
        self._tools = {}
        with open(config_path, encoding="utf-8") as f:
            self._config = json.load(f)

    def register_all(self):
        """按 config.json 的 tools 列表加载所有工具"""
        for item in self._config.get("tools", []):
            mod = importlib.import_module(item["module"])   # 按路径导入模块
            cls = getattr(mod, item["class"])                # 取类
            tool = cls(**self._deps)                        # 实例化（注入依赖）
            self._tools[tool.name] = tool
        return self

    def get_schemas(self): 
        """给 LLM 的说明书，返回 openai 可读的格式"""
        return [{"type": "function", "function": {
            "name": t.name, "description": t.description, "parameters": t.parameters,
        }} for t in self._tools.values()]

    def execute(self, name: str, args: dict) -> str:
        if name not in self._tools:
            return f"未注册的工具: {name}"
        return self._tools[name].execute(**args)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 出错重试 / 参数修复（跨工具通用）
from __future__ import annotations

import time

class ToolExecutor:
    def __init__(self, registry:ToolRegistry, max_retries=2):
        self._registry = registry
        self._max_retries = max_retries

    def execute_with_healing(self, name: str, args: dict) -> str:
        last_err = None  # 记录"最后一次错误"，备用
        for attempt in range(self._max_retries + 1):
            try:
                return self._registry.execute(name, args)   # 试执行工具
            except TypeError as e:                     # 模型给错参数键，修复重试
                args = self._repair_args(name, args, e) # 修复参数再试
                last_err = e
            except Exception as e:                    # 其他异常 → 重试
                last_err = e
                time.sleep(1)
        return f"[自愈失败] 工具 {name} 重试 {self._max_retries} 次仍失败: {last_err}"

    def _repair_args(self, name, args, err):
        """把模型给的别名参数映射回工具要的键名"""
        alias = {"user_fact": "fact", "preference": "fact",
                 "question": "query", "keyword": "query"}
        return {alias.get(k, k): v for k, v in args.items()}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# tools/weather.py（节选核心）
class WeatherTool(BaseTool):
    name = "get_weather"
    description = "查天气（按经纬度），返回天气、温度、湿度、日出日落、降雨概率"
    parameters = {
        "type": "object",
        "properties": {
            "latitude": {"type": "number", "description": "纬度"},
            "longitude": {"type": "number", "description": "经度"},
        },
        "required": ["latitude", "longitude"],
    }
    def execute(self, latitude: float, longitude: float) -> str:
        try:
            url = (f"https://api.open-meteo.com/v1/forecast?latitude={latitude}"
                   f"&longitude={longitude}&current=...&daily=...&timezone=Asia%2FShanghai")
            data = json.loads(urllib.request.urlopen(url).read())
            # ... 把 weather_code 翻成中文描述，拼成 JSON 返回
        except Exception as e:
            return f"查天气失败: {e}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# tools/calculator.py（节选核心）
class CalculatorTool(BaseTool):
    name = "calculator"
    description = "四则运算（如 1+2*3）"
    parameters = {
        "type": "object",
        "properties": {"expr": {"type": "string", "description": "数学表达式"}},
        "required": ["expr"],
    }
    def execute(self, expr: str) -> str:
        try:
            tree = ast.parse(expr, mode="eval")
            # 只允许数字和四则运算，防止模型注入危险代码
            allowed = (ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant,
                       ast.Add, ast.Sub, ast.Mult, ast.Div, ast.USub)
            for node in ast.walk(tree):
                if not isinstance(node, allowed):
                    return f"表达式包含不允许的语法: {type(node).__name__}"
            result = eval(compile(tree, "<string>", "eval"),
                          {"__builtins__": {}}, {})
            return str(result)
        except Exception as e:
            return f"计算失败: {e}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# tools/save_preference.py
class SavePreferenceTool(BaseTool):
    name = "save_preference"
    description = "当用户表达稳定偏好/事实时调用，跨会话记住"
    parameters = {
        "type": "object",
        "properties": {"fact": {"type": "string", "description": "要记住的用户事实"}},
        "required": ["fact"],
    }
    def __init__(self, mm=None):
        self.mm = mm          # 依赖注入：MemoryManager 由外面传入
    def execute(self, fact: str) -> str:
        if self.mm is None:
            return "错误：记忆管理器未注入"
        self.mm.add({"role": "user", "content": fact}, persist=True)  # 落长期
        return f"已保存到长期记忆 ✓"

# tools/recall_memory.py
class RecallMemoryTool(BaseTool):
    name = "recall_memory"
    description = "需要回顾用户历史偏好/事实时调用"
    parameters = {
        "type": "object",
        "properties": {"query": {"type": "string", "description": "检索关键词"}},
        "required": ["query"],
    }
    def __init__(self, mm=None):
        self.mm = mm
    def execute(self, query: str) -> str:
        if self.mm is None:
            return "错误：记忆管理器未注入"
        facts = self.mm.recall(query, k=3)          # 召回结果拼回上下文
        return "；".join(facts) if facts else "（没有找到相关记忆）"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import json
from .modes import BUILDERS

def fc_loop(llm, messages, registry, executor, max_rounds=3):
    """
    Function Calling 主循环
    1. llm.chat_for_tools 返回 message 对象 → 能拿到 tool_calls
    2. registry.get_schemas() 从 config.json 动态生成 schema
    3. executor.execute_with_healing 带自愈
    """
    tool_records = []
    for _ in range(max_rounds):
        msg = llm.chat_for_tools(messages, registry.get_schemas())
        if not msg.tool_calls:
            # 模型没调工具 → 返回答案 + 记录
            return {"answer": msg.content, "tool_records": tool_records}
        messages.append(msg)
        for tc in msg.tool_calls:
            try:
                args = json.loads(tc.function.arguments)
            except Exception as e:
                messages.append({"role": "tool", "tool_call_id": tc.id,
                                 "content": f"参数解析失败:{e}"})
                continue
            result = executor.execute_with_healing(tc.function.name, args)  # 自愈执行
            tool_records.append({"tool": tc.function.name, "args": args, "result": result})
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
    return {"answer": "达到最大工具轮数", "tool_records": tool_records}

def get_response(msgs, llm, **kw):
    """调 LLM 生成回复"""
    return llm.chat(msgs, **kw)

def run_mode(mode, question, llm, builders=None, **kw):
    """按模式调度：查注册表 → 构造消息 → 调 LLM"""
    builders = builders or BUILDERS
    if mode not in builders:
        raise ValueError(f"未知模式:{mode},可选 {list(builders)}")
    msgs = builders[mode](question, **kw)
    return get_response(msgs, llm, **kw)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# 路由：选模型 + 选推理模式
def route_model(question: str, default: str = "deepseek") -> str:
    """从用户输入里找模型名,找不到用默认"""
    for name in ["kimi", "deepseek", "qwen", "gpt"]:
        if name in question.lower():
            return name
    return default

def auto_select(question: str, llm=None) -> str:
    """
    判断问题适合哪种推理模式
    升级可选：让 LLM 输出 JSON 再解析，比关键词匹配更稳（防格式漂移）
    """
    if llm is not None:
        text = (f"判断下面用户的问题适合哪种回答，只输出以下之一：cot / few_shot / react / tot。"
                f"规则：需要动手查/调工具用 react；需要多方案权衡用 tot；"
                f"需要解释原因用 cot；需要举例对比用 few_shot。\n问题:{question}")
        out = llm.chat([{"role": "user", "content": text}]).lower()
        for mode in ["tot", "react", "few_shot", "cot"]:
            if mode in out:
                return mode

    q = question.lower()
    if any(k in q for k in ["帮我", "执行", "查询", "搜索"]):
        return "react"
    if any(k in q for k in ["如果", "方案", "计划"]):
        return "tot"
    if any(k in q for k in ["比较", "对比", "哪个"]):
        return "few_shot"
    return "cot"   # 默认

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

# agent/modes/cot.py —— 思维链：让模型一步步推
def cot_prompt(question: str) -> list[dict]:
    return [
        {"role": "system", "content": "你是擅长逐步推理的专家,请一步一步思考,先列思路再给结论。"},
        {"role": "user", "content": question},
    ]

# agent/modes/few_shot.py —— 少样本：给例子让它学格式
DEFAULT_EXAMPLES = [
    {"question": "我明天下午3点约了张医生复诊",
     "answer": "时间:明天下午3点;人物:张医生;事项:复诊"},
    {"question": "周五晚上和李总在望江楼吃饭",
     "answer": "时间:周五晚上;人物:李总;事项:吃饭"},
]
def few_shot_prompt(question, examples=DEFAULT_EXAMPLES,
                    system_prompt="你是擅长信息抽取的助手"):
    msgs = [{"role": "system", "content": system_prompt}]
    for ex in examples:
        msgs.append({"role": "user", "content": ex["question"]})
        msgs.append({"role": "assistant", "content": ex["answer"]})
    msgs.append({"role": "user", "content": question})
    return msgs

# agent/modes/react.py —— 工具调用：给个系统提示就完事，真正调工具靠 fc_loop
def react_prompt(question, system_prompt="你是会调用工具的智能助手，"
                 "需要查询/计算/记忆时调用工具，并把工具结果结合进回答。"):
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question},
    ]

# agent/modes/tot.py —— 树状思维：三次调用（发散→评审→决策）
def ask(llm, question, system_prompt="你是擅长多种角度对比方案的资深决策专家",
        max_tokens=500):
    msgs = [{"role": "system", "content": system_prompt},
            {"role": "user", "content": question}]
    return llm.chat(msgs, max_tokens=max_tokens)

def tot_solve(question, n=3, llm=None):
    """三段式：发散 → 评审 → 决策（三次调用）"""
    if llm is None:
        raise ValueError("tot 模式需要注入 llm")
    schemes = ask(llm, f"请针对以下问题给出{n}种不同解决思路,编号列出:\n{question}\n",
                  system_prompt="你是擅长发散思维的规划专家", max_tokens=600)
    scores = ask(llm, f"请对以下 {n} 个方案根据实行难度,风险评估逐一打分(1-10分),并各用一句话说明理由:\n{schemes}",
                 system_prompt="你是严格的评审专家,打分要客观", max_tokens=500)
    best = ask(llm, f"综合以下评分,选出最优方案,并给出具体实施步骤:\n{scores}",
               system_prompt="你是决策专家,直接给最终选择", max_tokens=600)
    return schemes, scores, best

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# main.py —— 组装所有层
import sys
from pathlib import Path

# 让 agent/main.py 直接运行时也能找到项目根目录的 llm/memory/tools 包
PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from llm import RealLLM
from memory import MemoryManager
from tools import ToolRegistry, ToolExecutor
from agent import fc_loop, auto_select

llm = RealLLM("deepseek")
mm = MemoryManager(window=8, max_tokens=1500, keep=4, llm=llm)
registry = ToolRegistry(str(PROJECT_ROOT / "config.json"), deps={"mm": mm}).register_all()
executor = ToolExecutor(registry)

while True:
    u = input("你: ")
    if u.lower() == "exit":
        break
    mm.add({"role": "user", "content": u}, persist=u.startswith("记住:"))
    ctx = mm.get_context(query=u)
    out = fc_loop(llm, ctx, registry, executor)   # 返回 {"answer", "tool_records"}
    mm.add({"role": "assistant", "content": out["answer"]})
    mm.maybe_compress()
    print("Agent:", out["answer"])