能否单独出个cli版

[lipeiying032]

我觉得gui太吃资源了，不仅是编译还是运行，我觉得可以单独出个低配置小主机适配的tui版，这样就可以在配置有限的vps也可以运行deepseek harness了

[AIxiaolong275]

顶这个吧
#172

[xw901103]

自己搓，我给你一个简单的例子，这个是我很早之前用的，很不安全但能用

import sys
import io
import os
import json
import traceback
import argparse
from datetime import datetime

import json5
from openai import OpenAI
from prompt_toolkit import PromptSession
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.lexers import PygmentsLexer
from prompt_toolkit.styles import Style
from prompt_toolkit.formatted_text import HTML
from pygments.lexers.python import PythonLexer

# ------------------------------------------------------------
# Configuration
# ------------------------------------------------------------
DEEPSEEK_API_KEY = "sk-your_key_here"
DEEPSEEK_MODEL = "deepseek-v4-pro"
IS_AGENT_RUNNING = True

# ------------------------------------------------------------
# Tool: run_python_code
# ------------------------------------------------------------
def tool_run_python_code(code: str, **kwargs) -> str:
    print("[code]")
    print(code)
    print("[/code]")

    safe_globals = {
        '__builtins__': __builtins__,
        'print': print,
    }

    old_stdout = sys.stdout
    old_stderr = sys.stderr
    sys.stdout = io.StringIO()
    sys.stderr = io.StringIO()

    try:
        exec(code, safe_globals)
        output = sys.stdout.getvalue()
        error = sys.stderr.getvalue()
    except Exception:
        output = ""
        error = traceback.format_exc()
    finally:
        sys.stdout = old_stdout
        sys.stderr = old_stderr

    return json5.dumps({
        "output": output,
        "error": error
    }, ensure_ascii=False)

# ------------------------------------------------------------
# Tools definition
# ------------------------------------------------------------
tools = [
    {
        "type": "function",
        "function": {
            "name": "run_python_code",
            "description": "Execute Python code and return its output.",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {
                        "type": "string",
                        "description": "The Python code to be executed",
                    }
                },
                "required": ["code"]
            },
        }
    },
]

tools_map = {
    "run_python_code": tool_run_python_code
}

# ------------------------------------------------------------
# Agent runner (unchanged core logic)
# ------------------------------------------------------------
def run_agent(client, messages, tools, tools_map):
    while True:
        stream = client.chat.completions.create(
            model=DEEPSEEK_MODEL,
            stream=True,
            messages=messages,
            tools=tools,
            tool_choice="auto",
            extra_body={
                "thinking": {"type": "enabled"},
                "reasoning_effort": "max"
            }
        )

        reasoning_chunks = []
        content_chunks = []
        tool_calls_dict = {}

        in_reasoning = False
        in_response = False

        for chunk in stream:
            delta = chunk.choices[0].delta
            finish_reason = chunk.choices[0].finish_reason

            if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
                if not in_reasoning:
                    print("[reasoning]", flush=True)
                    in_reasoning = True
                print(delta.reasoning_content, end='', flush=True)
                reasoning_chunks.append(delta.reasoning_content)

            if hasattr(delta, 'content') and delta.content:
                if in_reasoning:
                    print("\n[/reasoning]", flush=True)
                    in_reasoning = False
                if not in_response:
                    print("[response]", flush=True)
                    in_response = True
                print(delta.content, end='', flush=True)
                content_chunks.append(delta.content)

            if hasattr(delta, 'tool_calls') and delta.tool_calls is not None:
                for tc_delta in delta.tool_calls:
                    idx = tc_delta.index
                    if idx not in tool_calls_dict:
                        tool_calls_dict[idx] = {
                            'id': None,
                            'name': None,
                            'arguments_chunks': []
                        }
                    if tc_delta.id:
                        tool_calls_dict[idx]['id'] = tc_delta.id
                    if tc_delta.function:
                        if tc_delta.function.name:
                            tool_calls_dict[idx]['name'] = tc_delta.function.name
                        if tc_delta.function.arguments:
                            tool_calls_dict[idx]['arguments_chunks'].append(
                                tc_delta.function.arguments
                            )

            if finish_reason is not None:
                break

        if in_reasoning:
            print("\n[/reasoning]", flush=True)
        if in_response:
            print("\n[/response]", flush=True)

        reasoning_content = ''.join(reasoning_chunks) if reasoning_chunks else None
        content = ''.join(content_chunks) if content_chunks else None
        tool_calls = None

        if tool_calls_dict:
            tool_calls = []
            for idx in sorted(tool_calls_dict.keys()):
                tc = tool_calls_dict[idx]
                tool_calls.append({
                    'id': tc['id'],
                    'type': 'function',
                    'function': {
                        'name': tc['name'],
                        'arguments': ''.join(tc['arguments_chunks'])
                    }
                })

        messages.append({
            "role": "assistant",
            "reasoning_content": reasoning_content,
            "content": content,
            "tool_calls": tool_calls
        })

        if tool_calls is None:
            break

        print("[tools]")
        for tool in tool_calls:
            function_name = tool['function']['name']
            tool_function = tools_map[function_name]
            tool_result = tool_function(**json5.loads(tool['function']['arguments']))
            print(f"[{function_name}]\n{tool_result}\n[/{function_name}]")
            messages.append({
                "role": "tool",
                "tool_call_id": tool['id'],
                "content": tool_result,
            })
        print("[/tools]")

# ------------------------------------------------------------
# Helper: Remove last assistant reply (and subsequent tool messages)
# ------------------------------------------------------------
def remove_last_assistant_reply(messages):
    """
    Remove the last assistant message and all subsequent tool messages
    from the messages list. Returns True if something was removed, False otherwise.
    """
    # Find the index of the last user message
    last_user_idx = None
    for i in range(len(messages) - 1, -1, -1):
        if messages[i].get('role') == 'user':
            last_user_idx = i
            break

    if last_user_idx is None:
        # No user message found, nothing to regenerate
        print("[regenerate] No user message found; nothing to regenerate.")
        return False

    # Check if there is at least one assistant message after the last user message
    has_assistant_after = False
    for i in range(last_user_idx + 1, len(messages)):
        if messages[i].get('role') == 'assistant':
            has_assistant_after = True
            break

    if not has_assistant_after:
        print("[regenerate] No assistant reply after the last user message; nothing to regenerate.")
        return False

    # Truncate: keep everything up to and including the last user message
    removed_count = len(messages) - (last_user_idx + 1)
    del messages[last_user_idx + 1:]
    print(f"[regenerate] Removed {removed_count} message(s) (assistant reply + any tool results).")
    print("[regenerate] Ready to regenerate the last reply.")
    return True


# ------------------------------------------------------------
# JSON session file helpers
# ------------------------------------------------------------
def load_messages_from_json(filepath):
    """Load messages from a JSON file. Returns list of messages."""
    if not os.path.exists(filepath):
        print(f"[Warning] File not found: {filepath}. Starting a new session.")
        return None
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        # Support both a list of messages or a dict with a "messages" key
        if isinstance(data, list):
            messages = data
        elif isinstance(data, dict) and "messages" in data:
            messages = data["messages"]
        else:
            print(f"[Warning] Unrecognized format in {filepath}. Starting a new session.")
            return None
        print(f"[Loaded {len(messages)} messages from {filepath}]")
        return messages
    except (json.JSONDecodeError, IOError) as e:
        print(f"[Error] Failed to load {filepath}: {e}. Starting a new session.")
        return None

def save_messages_to_json(filepath, messages):
    """Save messages to a JSON file."""
    try:
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(messages, f, ensure_ascii=False, indent=2)
        # Silently save; we can print a message if needed
    except IOError as e:
        print(f"[Error] Failed to save session to {filepath}: {e}")

def generate_timestamp_filename():
    """Generate a filename based on the current timestamp."""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    return f"ds-agent-session-{timestamp}.json"

# ------------------------------------------------------------
# prompt_toolkit setup
# ------------------------------------------------------------

# Custom key bindings: Enter submits only when pressed with Alt;
# plain Enter inserts a newline (multi-line mode).
bindings = KeyBindings()

@bindings.add('enter')
def handle_enter(event):
    """Insert a newline on plain Enter (multi-line)."""
    event.current_buffer.insert_text('\n')

@bindings.add('escape', 'enter')
def handle_escape_enter(event):
    """Submit on Esc + Enter."""
    event.current_buffer.validate_and_handle()

@bindings.add('c-d')
def handle_ctrl_d(event):
    """Exit on Ctrl+D if the input buffer is empty."""
    buffer = event.current_buffer
    if buffer.text.strip() == '':
        event.app.exit(result=None)
    else:
        # If there's text, delete forward (default behavior fallback)
        buffer.delete()

# Style for the prompt
style = Style.from_dict({
    'prompt': 'ansicyan bold',
    'toolbar': 'bg:#333333 #ffffff',
    'separator': '#666666',
})

def get_toolbar():
    """Return the bottom toolbar text."""
    return HTML(
        ' <b>Multi-line mode</b> | '
        '<ansigreen>Enter</ansigreen> = newline | '
        '<ansigreen>Esc+Enter</ansigreen> = submit | '
        '<ansigreen>Ctrl+D</ansigreen> = quit (empty input) | '
        '<ansigreen>/regen</ansigreen> = regenerate last | '
        '<ansigreen>/cont</ansigreen> = continue | '
        '<ansigreen>/help</ansigreen> = help'
    )

# Create the PromptSession with history
session = PromptSession(
    lexer=PygmentsLexer(PythonLexer),   # syntax highlighting for Python
    style=style,
    key_bindings=bindings,
    multiline=True,
    bottom_toolbar=get_toolbar,
    history=None,  # you can pass a FileHistory object for persistence
)

# ------------------------------------------------------------
# Client initialization
# ------------------------------------------------------------
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com",
)

# ------------------------------------------------------------
# Main loop
# ------------------------------------------------------------
if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="DeepSeek Agent with prompt_toolkit – multi-line interactive chat"
    )
    parser.add_argument(
        "json_file",
        nargs="?",
        default=None,
        help="Path to a JSON message record file. If not provided, a new file "
             "with a timestamp is created automatically."
    )
    parser.add_argument(
        "--new", "-n",
        action="store_true",
        help="Start a fresh session even if the JSON file exists (ignores existing messages)."
    )
    args = parser.parse_args()

    # Determine the JSON file path
    if args.json_file:
        json_filepath = args.json_file
    else:
        json_filepath = generate_timestamp_filename()
        print(f"[No JSON file provided. Creating new session file: {json_filepath}]")

    # Load or initialize messages
    messages = None
    if args.new:
        print("[Starting a brand-new session (--new flag)]")
        messages = [{'role': 'system', 'content': 'You are a helpful assistant.'}]
    elif args.json_file and os.path.exists(args.json_file):
        loaded = load_messages_from_json(json_filepath)
        if loaded is not None:
            messages = loaded
            # Ensure the system message exists; if not, prepend one
            if not any(m.get('role') == 'system' for m in messages):
                messages.insert(0, {'role': 'system', 'content': 'You are a helpful assistant.'})
        else:
            messages = [{'role': 'system', 'content': 'You are a helpful assistant.'}]
    else:
        messages = [{'role': 'system', 'content': 'You are a helpful assistant.'}]

    # Save the initial (or loaded) state
    save_messages_to_json(json_filepath, messages)

    print("=" * 60)
    print(" DeepSeek Agent – Multi-line Prompt ")
    print("=" * 60)
    print(f" Session file: {json_filepath}")
    print(" Type your message. Use Esc+Enter to send, Ctrl+D to exit.")
    print(" Commands:")
    print("   /regen   – delete last assistant reply and regenerate it")
    print("   /cont    – continue generating (submit conversation as-is)")
    print("   /help    – show this help")
    print()

    while IS_AGENT_RUNNING:
        try:
            user_input = session.prompt(
                HTML('<ansicyan><b>User&gt;</b></ansicyan> '),
                multiline=True,
            )
        except KeyboardInterrupt:
            # Ctrl+C pressed
            print("\n[Interrupted]")
            continue
        except EOFError:
            # Ctrl+D on empty input → exit
            print("\n[Exiting]")
            break

        if user_input is None:
            break

        user_input = user_input.strip()
        if not user_input:
            continue

        if user_input.lower() in ("quit", "exit"):
            IS_AGENT_RUNNING = False
            continue

        # --- Command handling ---
        if user_input.startswith('/'):
            cmd = user_input[1:].strip().lower()

            if cmd in ('regen', 'regenerate'):
                # Regenerate the last assistant reply
                if remove_last_assistant_reply(messages):
                    # Now run the agent again with the truncated messages
                    run_agent(client, messages, tools, tools_map)
                    save_messages_to_json(json_filepath, messages)
                else:
                    # Nothing to regenerate; maybe just run the agent (treat as /cont)
                    print("[regenerate] No reply to regenerate. Treating as /cont (continue).")
                    run_agent(client, messages, tools, tools_map)
                    save_messages_to_json(json_filepath, messages)
                continue

            elif cmd in ('cont', 'continue'):
                # Continue generating: submit the conversation as-is
                print("[continue] Submitting conversation to model for continuation...")
                run_agent(client, messages, tools, tools_map)
                save_messages_to_json(json_filepath, messages)
                continue

            elif cmd in ('help', '?'):
                print()
                print("  Available commands:")
                print("    /regen   – delete last assistant reply and regenerate it")
                print("    /cont    – continue generating (submit conversation as-is)")
                print("    /help    – show this help")
                print("    quit/exit – exit the program")
                print()
                continue

            else:
                print(f"[Unknown command: /{cmd}. Type /help for available commands.]")
                continue

        # Normal user message: append and run agent
        messages.append({'role': 'user', 'content': user_input})
        run_agent(client, messages, tools, tools_map)

        # Save after each complete turn (user + assistant + possible tool calls)
        save_messages_to_json(json_filepath, messages)

    print("Goodbye!")

[bobleer]

我在用 Rust 重写 harness：deepseek-harness-rust，分层 crates（session log / turn-step loop / DeepSeek SSE adapter），CLI 可以在这个底子上出，headless 跑起来很轻。github.com/bobleer/deepseek-harness-rust，MIT，欢迎共建。

[AIxiaolong275]

等官方吧，你们这么着急

[tianxiongWang]

顶上去

[Electricitysheep]

补充一个当前可用的方案：headless profile 就是"命令行形态"——dsh --profile headless "任务" 适合脚本/CI（打印结果退出、非零码即失败）。我们已经实测用它在 CI 场景跑数据任务、代码重构（见 dsh-handbook https://github.com/Electricitysheep/dsh-handbook 第 2/5 章，含真实耗时数据）。

交互式 CLI/TUI 目前需要插件（官方未内置）——这确实是个缺口，支持官方早日补上。

[gxinxing]

https://github.com/gxinxing/deepseek-harness-tui 看看

[weijiafu14]

低配 VPS / 纯 SSH 的场景现在可以走 @deepseek-harness-tui/dsh-tui，不用启动 Web 界面。想把 Pi 生态插件一并带进 TUI，再安装 pi2dsh：

dsh plugin --profile dsh-tui add @deepseek-harness-tui/dsh-tui
dsh plugin --profile dsh-tui add pi2dsh

我们用 npm 原版 DSH、原版 dsh-TUI 跑过真实模型、两个独立 Agent 以及插件的命令和工具链；没有另造 Agent loop。步骤和证据：https://github.com/weijiafu14/pi2dsh#walkthrough-advanced-mcp-in-your-terminal