127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Bridge script: caller agent → novelist agent (via Gateway API)
|
|
|
|
Automatically copies the novel source text to the novelist's workspace so that
|
|
the novelist agent can read the actual body text locally, not just chapter titles.
|
|
Then sends the request via Gateway API synchronously.
|
|
|
|
Usage:
|
|
python ask_novelist.py "continuation request"
|
|
python ask_novelist.py --novel /path/to/novel.txt "continuation request"
|
|
cat content.txt | python ask_novelist.py
|
|
"""
|
|
import sys
|
|
import os
|
|
import shutil
|
|
import json
|
|
import requests
|
|
|
|
# Config — customize these for your setup
|
|
GATEWAY_URL = "http://127.0.0.1:18789/v1/chat/completions"
|
|
GATEWAY_TOKEN = "<your-gateway-token>"
|
|
NOVELIST_MODEL = "openclaw/novelist"
|
|
NOVELIST_WORKSPACE = os.environ.get("NOVELIST_WORKSPACE", "/home/program/novelist-workspace")
|
|
NOVEL_TARGET_NAME = "novel_source.txt"
|
|
|
|
|
|
def ensure_novel_in_workspace(novel_path: str = None) -> str | None:
|
|
"""Copy the novel file to the novelist's workspace for local reading."""
|
|
if not novel_path or not os.path.isfile(novel_path):
|
|
# Try common default paths
|
|
candidates = [
|
|
"./novel.txt",
|
|
os.path.expanduser("~/novel.txt"),
|
|
]
|
|
for c in candidates:
|
|
if os.path.isfile(c):
|
|
novel_path = c
|
|
break
|
|
else:
|
|
return None
|
|
|
|
target = os.path.join(NOVELIST_WORKSPACE, NOVEL_TARGET_NAME)
|
|
try:
|
|
if not os.path.isfile(target) or os.path.getsize(target) != os.path.getsize(novel_path):
|
|
shutil.copy2(novel_path, target)
|
|
print(f"[桥接] 已复制小说到 {target}", file=sys.stderr)
|
|
return target
|
|
except Exception as e:
|
|
print(f"[警告] 复制小说文件失败: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def ask(question: str, timeout: int = 180) -> str:
|
|
"""Synchronous Gateway API call to novelist."""
|
|
payload = {
|
|
"model": NOVELIST_MODEL,
|
|
"messages": [{"role": "user", "content": question}],
|
|
"stream": False,
|
|
}
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {GATEWAY_TOKEN}",
|
|
}
|
|
try:
|
|
resp = requests.post(GATEWAY_URL, json=payload, headers=headers, timeout=timeout)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
choices = data.get("choices", [])
|
|
if choices:
|
|
return choices[0].get("message", {}).get("content", "")
|
|
return "[novelist] 返回为空"
|
|
return f"[novelist] 请求失败({resp.status_code}): {resp.text[:200]}"
|
|
except requests.exceptions.Timeout:
|
|
return "[novelist] 超时，请稍后重试"
|
|
except requests.exceptions.ConnectionError:
|
|
return "[novelist] 连接失败，请检查 Gateway 服务状态"
|
|
except Exception as e:
|
|
return f"[novelist] 请求异常: {str(e)}"
|
|
|
|
|
|
def build_prompt(novel_path: str | None, user_question: str) -> str:
|
|
"""Build the prompt sent to novelist. Includes novel file path hint."""
|
|
if novel_path:
|
|
return (
|
|
f"小说原文已放在你的工作区，路径为：{novel_path}\n\n"
|
|
f"请先阅读原文正文（不要只看标题），完成要求。\n\n"
|
|
f"要求如下：\n{user_question}"
|
|
)
|
|
return user_question
|
|
|
|
|
|
def main():
|
|
novel_path = None
|
|
args = sys.argv[1:]
|
|
|
|
if args and args[0] == "--novel":
|
|
if len(args) >= 2:
|
|
novel_path = args[1]
|
|
question = " ".join(args[2:]) if len(args) > 2 else ""
|
|
else:
|
|
print("错误：--novel 需要提供文件路径", file=sys.stderr)
|
|
sys.exit(1)
|
|
elif args:
|
|
question = " ".join(args)
|
|
|
|
if not question:
|
|
question = sys.stdin.read().strip()
|
|
|
|
if not question:
|
|
print("错误：未提供请求内容", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
copied_path = ensure_novel_in_workspace(novel_path)
|
|
if copied_path:
|
|
novel_file = os.path.join(NOVELIST_WORKSPACE, NOVEL_TARGET_NAME)
|
|
else:
|
|
novel_file = None
|
|
|
|
prompt = build_prompt(novel_file, question)
|
|
result = ask(prompt)
|
|
print(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|