|
tpf 是自研工具库，LLM 部分分两层：
```text
tpf/
├── llm/ # 底层 LLM 网关
│ ├── openai.py # chat() 函数 + MyChat 类（多厂商门面）+ 文生图
│ ├── ollama.py # client_ollama / chat_ollama / chat（本地模型）
│ ├── funcall.py # FuncCall 类（函数调用双引擎）
│ ├── tools.py # Tools 类（JSON 解析 + 工具定义）
│ ├── prompt.py # return_json1 输出契约 + SqlJiaoYi nl2sql 提示工厂
│ └── agent.py # Agent 类（Agently 框架封装）
├── cn/ # 高层"超级大脑"
│ ├── link.py # Think 类（对话 + 函数自动生成）
│ ├── prompt.py # Prompt 类（需求分析/代码生成提示模板）
│ ├── funcs.py # 函数工具库 + _register_function 自动注册
│ └── tools/windows.py # BatchGenerator（cmd转bat，单例）
└── funcs/base.py # append_code_to_file_safely 等通用函数
```
公开入口：`from tpf.llm import MyChat, chat, FuncCall, Tools`；`from tpf.cn import Think, BatchGenerator`。
|
|
**全库统一 JSON 返回契约**（`return_json1()`，所有 `return_json=True` 的调用都注入该约束）：
```json
{
"query": "用户问题",
"result": "最终回复（代码生成场景为纯 Python 代码）",
"thinking": ["思考步骤1", "思考步骤2"]
}
```
解析链：模型输出含 ```` ```json ... ``` ```` 围栏 → `Tools.get_json_str()` 用正则 ```` ```json([\s\S]*?)``` ```` 剥离 → `json.loads`；**解析失败优雅降级返回原文**，不抛异常。
|
|
env 文件（默认 `.env`，回退 `env.txt` → `/wks/app/conf/env.txt`）：
```text
deepseek_api_key=sk-xxx
deepseek_base_url=https://api.deepseek.com/v1
OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=https://.../v1
DASHSCOPE_API_KEY=sk-xxx # 通义
qianfan_ak=xxx / qianfan_sk=xxx # 千帆
ARK_API_KEY / ARK_BASE_URL # 火山方舟（豆包）
glm_api_key / glm_base_url # 智谱
API_KEY / BASE_URL # 通用 request 裸调用
```
命名规范：`{厂商}_api_key` 与 `{厂商}_base_url`。
|
|
最底层 OpenAI 兼容对话函数（MyChat 各方法都经它转发）：
```python
from tpf.llm import chat
content = chat(
prompt_user="你好",
prompt_system=None,
response_format="text", # 传给 API 的 {"type": ...}
model='deepseek-r1:1.5b',
temperature=1,
base_url='http://localhost:11434/v1/', # 默认走本地 ollama
api_key='key',
return_json=False # True 时注入 JSON 契约并解析
)
```
本地 ollama 也有专用封装：
```python
from tpf.llm.ollama import chat_ollama
res = chat_ollama("你好", model='deepseek-r1:1.5b', temperature=0)
```
|
|
MyChat 统一入口，一族方法打天下：
```python
from tpf.llm import MyChat
chat = MyChat(env_file="env.txt")
# 对话类：deepseek / openai / tongyi / qianfan / ollama
res = chat.deepseek(prompt_user="你好", models=['deepseek-chat','deepseek-reasoner'],
temperature=0, return_json=True)
print(res['result'], res['thinking'])
res = chat.ollama(prompt_user="你好",
model=["DeepSeek-R1-32B-Q8:latest"], # 列表=多备选
temperature=1)
# 用 openai 方法走任意 OpenAI 兼容端点（例：deepseek）
res = chat.openai(prompt_user="你好", models='deepseek-chat', temperature=0,
base_url="https://api.deepseek.com/v1", api_key="...", return_json=True)
# 函数调用类：*_func_call（见 FuncCall 页）
res = chat.deepseek_func_call(query="请启动dcn服务", model_list=[...], func_index=0)
# 本地模型注册（func_call 路由用）
chat.set_local_model("DeepSeek-R1-14B-Q8:latest")
```
`set_local_model/get_local_model` 维护本地 ollama 名单——决定 func_call 走"本地自定义流程"还是"在线 tools 流程"。
|
|
三家文生图统一返回 `(url, img_save_path)`，自动下载落盘：
```python
# 智谱
url, path = chat.glm_image("一张雪山的海报", outdir="",
models=["glm-image","cogview-3"], size="1024x1024")
# 通义（支持 negative_prompt / prompt_extend / watermark）
url, path = chat.tongyi_image("一只柴犬", models=["wan2.6-t2i","qwen-image-max"],
size="1024*1024", watermark=False)
# 豆包（ARK，OpenAI SDK images.generate）
url, path = chat.doubao_image("赛博朋克城市夜景", size="2K")
```
实现要点：重试递增等待（max_retries=3）、120s 超时、文件名 `glm_时间戳.png`。
|
|
`request()` 方法做裸 HTTP 调用（URL 自动补 `/v1/chat/completions`，`**params` 直接并入请求体，支持 n/stream/presence_penalty/frequency_penalty/max_tokens/top_p）：
```python
content = chat.request(
prompt_user="你好", models=["deepseek-chat"], temperature=0.1,
base_url=None, api_key=None,
top_p=0.9, max_tokens=1024
)
```
返回 `choices[0].message.content`，120 秒超时——适合调那些"OpenAI 兼容但 SDK 不认"的端点。
|
|
`tpf.cn.link.Think` 是高层统一对话入口：
```python
from tpf.cn import Think
thk = Think() # 默认读 .env，回退 env.txt
res = thk.thinking(
prompt_user="帮我分析这段需求的实现思路：...",
prompt_system=None,
response_format="text", # "json_object" 时配合 return_json
models=['deepseek-v4-flash','deepseek-reasoner','deepseek-chat'], # 只取第1个，列表便于换序
temperature=0.1,
base_url=None, api_key=None, # None 则用 env 里的 deepseek 配置
return_json=False, # True 时经 ```json``` 围栏解析为 dict
file=None # 指定则把结果写盘（json 用 ensure_ascii=False）
)
```
内部流程：默认参数补全 → OpenAI 客户端 → [system]+[user] 消息 → chat.completions → content → （可选）JSON 解析/写盘 → 返回。
通义封装：`thk.tongyi(prompt_user, models=['qwen3-coder-plus'], temperature=0, return_json=True)` 内部转调 thinking 换 DashScope 端点。
|
|
Think 的特色：**"分析→生成→注册→调用"函数自动生成链**：
```python
# 1. 需求分析：是否需要函数？要几个？
ana = thk.deepseek_func_needing(query)
# 返回 {"query","result":bool,"thinking":[(内容,是否需函数0/1,函数信息)],"tools_num":n}
# 2. 代码生成：按分析结果生成函数代码（纯代码、语法自检、尾部自动加注册行）
res = thk.deepseek_func_gen(query, func_path="/tmp/funcs.py")
# 内部: Prompt.gen_fun -> thinking(return_json=True) -> append_code_to_file_safely 落盘
print(res['result']) # 函数代码文本
```
生成的函数尾部带 `_register_function(函数名)`，写入 `cn/funcs.py` 即自动注册进工具库（装饰器从 docstring 首行取描述、从签名取参数生成 schema）。
再经 `MyChat.add_funcs(tools_info, tools_mapping)` 注册给 FuncCall，随后 `chat.deepseek_func_call(query)` 即可调度——端到端把"一句话需求"变成"可调用工具"。
注意：`Think.func_from_file_call` 引用的 `func_call_add_head` 在当前 funcs.py 中不存在（历史遗留），此路径会 ImportError，待修复。
|
|
BatchGenerator（单例）：把在 cmd/PowerShell 敲过的命令序列转成 .bat 脚本：
```python
from tpf.cn import BatchGenerator
bat = BatchGenerator().bat_gen(
"pip install pandas\ndel /f temp.log"
)
print(bat) # 生成完整 bat 脚本文本
```
实现要点：`__new__` 单例共享一个 Think；few-shot 用一对 ask/answer 示例教格式；输入经 `repr()` 转原始字符串避免反斜杠转义问题。
|
tpf 源码（aitpf 工程内）