Jan-V1：含演示项目的指南

Jan 团队近期发布了 Jan-v1，这是一款先进的代理式推理模型，通过强化的工具使用与多步问题求解，引入了成熟的研究自动化能力。

本教程将聚焦 Jan-v1 独特的代理式推理能力，展示它如何通过基于 llama.cpp 的本地部署和 Streamlit 界面，充当复杂的研究工作流。

在本教程中，您将学到：

• 使用 GGUF 量化在本地部署 Jan-v1，以实现性能与隐私的最佳平衡
• 构建用于自动化研究并实时跟踪进度的 Streamlit 应用
• 实现具有动态查询生成的智能网页搜索集成
• 创建具备特定格式结构的专业级报告综述
• 应用高级文本处理，确保文档干净、可直接用于业务

完成后，您的应用将如下所示：

什么是 Jan-v1？

Jan-v1 是一款拥有 40 亿参数的模型，基于 Lucy 和 Qwen3-4B-thinking 基座构建，专为工具调用、多步推理与自动化研究等代理式工作流而设计。与传统 LLM 不同，Jan-v1 擅长通过结构化脚手架与行为对齐奖励，在工具、数据源与推理步骤之间高效协同。

以下是该模型的一些关键特性：

• 面向代理的架构：该模型针对真实世界集成进行了优化，依托强化的推理与工具 API，处理网页搜索、信息综合与结构化报告生成。
• 高效性：尽管体量较小，Jan-v1 借助以目标导向输出为重心的定向 RL 训练，呈现出通常在更大型模型中才见到的表现。
• 隐私优先的部署：其精简的 GGUF 格式支持快速本地推理，非常适合在笔记本或 CPU 上进行低延迟、私密的研究任务。
• 面向研究的训练：Jan-v1 采用新的奖励函数，在信息量与简洁性间取得平衡，这对分析与报告等高要求任务至关重要，从而减少幻觉与冗长。
• 基于 Qwen3-4B-thinking：该模型继承了 Qwen3 的思维模式融合所带来的稳健链式推理与工具控制能力，对构建研究自动化流水线尤为有效。

在本地设置 Jan-v1

在本项目中，我们将使用 Llama.cpp 搭配 GGUF 量化版本在本地部署 Jan-v1，以获得最佳性能与资源效率。

步骤 1：安装依赖

首先，确保您已安装所需的 Python 包：

pip install llama-cpp-python==0.3.15
pip install streamlit aiohttp 技术支持

该命令将安装 AI 模型集成、网页界面构建与异步网络操作所需的核心依赖。

步骤 2：下载 Jan-v1 的 GGUF 版本

下载模型前，使用以下命令创建名为 models/ 的新目录：

 mkdir -p models技术支持

然后使用以下命令下载模型：

curl -L https://huggingface.co/janhq/Jan-v1-4B-GGUF/resolve/main/Jan-v1-4B-Q4_K_M.gguf -o Jan-v1-4B-Q4_K_M.gguf技术支持

完成此步骤后，您已成功下载 GGUF 格式的 Jan-v1 4B 模型，并将其放入 models/ 目录。该量化版本在性能与内存效率之间取得良好平衡，非常适合在 CPU 或 Colab 上进行本地推理。

步骤 3：配置 Serper API 密钥

注册或登录您的 Serper 账户： https://serper.dev/api-keys，选择 API Keys 选项卡，然后复制本演示所用的默认 API 密钥。将该密钥设置为环境变量如下：

export SERPER_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxxx技术支持

现在，您已准备好在研究助手流水线中加载并与模型交互。

演示：基于 Jan-v1 的深度研究助手

我们来构建一个 Streamlit 应用，演示 Jan-v1 的高级研究能力，包括智能查询生成、自动网页搜索（使用 Serper）以及专业报告综合。

步骤 1：导入库

首先导入研究助手所需的全部库，涵盖从网页界面构建到异步网页搜索与本地 AI 模型集成等功能。

import streamlit as st
import json
import time
import re
from datetime import datetime
from typing import List, Dict, Optional
import asyncio
import aiohttp
from llama_cpp import Llama
import os
from dataclasses import dataclass技术支持

我们导入了应用使用的核心库，包括 UI 组件（Streamlit）、并发工具（asyncio、aiohttp）以及 LLM 后端（llama_cpp）。同时引入了解析、类型标注与日期处理等实用工具。

步骤 2：辅助函数

接下来实现若干用于报告格式化与内容提取的辅助函数，以确保生成的报告具备专业质量与一致结构。

def sections_for_format(fmt: str) -> list[str]:
    fmt = (fmt or "").strip().lower()
    if fmt == "executive":
        return ["EXECUTIVE SUMMARY"]
    if fmt == "detailed":
        return [
            "INTRODUCTION",
            "DETAILED ANALYSIS",
            "CURRENT TRENDS AND DEVELOPMENTS",
            "IMPLICATIONS AND RECOMMENDATIONS",
            "CONCLUSION",
        ]
    if fmt == "academic":
        return [
            "ABSTRACT",
            "INTRODUCTION",
            "METHODOLOGY",
            "FINDINGS",
            "DISCUSSION",
            "CONCLUSION",
        ]
    if fmt == "presentation":
        return [
            "OVERVIEW",
            "KEY INSIGHTS",
            "RECOMMENDATIONS",
            "NEXT STEPS",
            "CONCLUSION",
        ]
    return ["INTRODUCTION", "DETAILED ANALYSIS", "CONCLUSION"]
def extract_final_block(text: str) -> str:
    m = re.search(r"<final>([\s\S]*?)</final>", text, flags=re.IGNORECASE)
    if m:
        cleaned_text = m.group(1).strip()
    else:
        cleaned_text = text
    preamble_patterns = [
        r"^(?:note:|okay,|hmm,|internal|let me|i (?:will|’ll)|as an ai|thinking|plan:|here is your report|the following is|i have prepared|i am presenting|based on the provided information|below is the report|i hope this meets your requirements|this report outlines|this is the final report).*?$",
        r"^(?:Here is the report|I have compiled the report|The report is provided below|This is the requested report).*?$", # More specific preambles
        r"^(?:Please find the report below|Here's the report).*?$"
    ]
    for pattern in preamble_patterns:
        cleaned_text = re.sub(pattern, "", cleaned_text, flags=re.IGNORECASE | re.MULTILINE).strip()
    cleaned_text = re.sub(r"(?m)^\s*[-*•]\s+", "", cleaned_text)
    cleaned_text = re.sub(r"[#`*_]{1,3}", "", cleaned_text)
    headers = [
        "EXECUTIVE SUMMARY","INTRODUCTION","DETAILED ANALYSIS","CURRENT TRENDS AND DEVELOPMENTS",
        "IMPLICATIONS AND RECOMMENDATIONS","CONCLUSION","ABSTRACT","METHODOLOGY","FINDINGS",
        "DISCUSSION","OVERVIEW","KEY INSIGHTS","RECOMMENDATIONS","NEXT STEPS"
    ]
    sorted_headers = sorted(headers, key=len, reverse=True)
    first_pos = -1
    for h in sorted_headers:
        match = re.search(r'\b' + re.escape(h) + r'\b', cleaned_text, flags=re.IGNORECASE)
        if match:
            if first_pos == -1 or match.start() < first_pos:
                first_pos = match.start()
    if first_pos >= 0:
        cleaned_text = cleaned_text[first_pos:].strip()
    return cleaned_text技术支持

各函数在流程中的作用如下：

• sections_for_format() 函数： 根据所选报告格式（如 executive、detailed、academic）生成相应的章节标题，确保每份报告符合专业标准并满足用户预期。
• extract_final_block() 函数： 该后处理步骤会提取核心报告内容：抓取 <final> 标签内的文本并移除开场白、冗余短语与 Markdown 痕迹；同时检测最早出现的标准标题（如 "INTRODUCTION"、"SUMMARY"），去除不相关的引导性文本，确保最终输出干净、结构清晰、可直接使用。

这些辅助函数对于在不同报告格式与研究场景下，保持像 Jan-v1 这类思维型模型输出的一致性与高质量至关重要。

步骤 3：配置设置

我们建立一个核心配置系统，用于管理模型参数、API 设置与研究偏好。该模块化方法便于自定义与性能优化。

步骤 3.1：研究配置

本步骤定义了一个配置系统和模型加载器，用于在仅 CPU 环境（如笔记本或 Colab Pro）中高效地对 Jan-v1 进行本地推理，同时保持灵活与安全。

@dataclass
class ResearchConfig:
    model_path: str = "models/Jan-v1-4B-Q4_K_M.gguf" #change as per your location
    max_tokens: int = 2048
    temperature: float = 0.6
    top_p: float = 0.95
    top_k: int = 20
    context_length: int = 4096
    search_api_key: str = os.getenv("SERPER_API_KEY", "")
    search_engine: str = "serper"  技术支持

上述代码的配置概览如下：

• 集中化参数：ResearchConfig 是一个 @dataclass，集中管理所有 LLM 与搜索引擎配置，便于在不同研究场景下进行调优。
• 推理超参数：temperature、top_p 与 top_k 等参数在创造性与事实性之间取得平衡，这对研究取向的输出尤为重要。
• 内存与上下文：context_length 设为 4096 token，优化以支持 Jan-v1 的长文本推理能力。
• API 集成：search_api_key 支持接入 Serper，用于实时网页搜索增强。

步骤 3.2：深度研究助手

本步骤定义一个深度研究助手类，作为所有研究任务的核心控制器。它抽象了底层模型与配置管理，同时提供用于生成、搜索与综合的高级接口。

class DeepResearchAssistant:
    def __init__(self, config: ResearchConfig):
        self.config = config
        self.llm = None
        self.demo_mode = False
    def load_model(self):
        try:
            if not os.path.exists(self.config.model_path):
                print(f"Model file not found: {self.config.model_path}")
                return False
            file_size_gb = os.path.getsize(self.config.model_path) / (1024**3)
            if file_size_gb < 1.0:
                print(f"Model file too small ({file_size_gb:.1f} GB).")
                return False
            self.llm = Llama(
                model_path=self.config.model_path,
                n_ctx=self.config.context_length,
                verbose=False,
                n_threads=max(1, min(4, os.cpu_count() // 2)),
                n_gpu_layers=0,
                use_mmap=True,
                use_mlock=False,
                n_batch=128,
                f16_kv=True,
            )
            test = self.llm("Hi", max_tokens=3, temperature=0.1, echo=False)
            ok = bool(test and 'choices' in test)
            print("Model loaded " if ok else "Model loaded but test generation failed ")
            return ok
        except Exception as e:
            print(f"Model loading failed: {e}")
            return False技术支持

DeepResearchAssistant 类是主要的编排器，负责管理模型加载、配置校验，并提供简洁的研究操作接口。该配置系统具备错误处理与校验，确保模型正确加载并可投入研究任务。其关键特性包括：

• 初始化模型： load_model() 方法通过 llama-cpp-python 安全初始化 Jan-v1 GGUF 模型，并针对本地 CPU 推理进行了优化。
• 校验检查：先验证模型路径与最小文件大小，以捕获缺失或损坏的文件。
• 内存映射：采用 use_mmap=True 启用高效的基于磁盘加载，无需占用全部内存。
• 推理优化：配置线程、批大小，并启用 f16_kv=True 以降低内存使用；n_gpu_layers=0 确保仅使用 CPU 执行。
• 健全性测试：加载后发送轻量提示进行测试，确认模型响应正常。

该设置确保了快速、轻量的本地推理，非常适合在无 GPU 的笔记本或 Colab 上运行 Jan-v1。

步骤 3.3：生成响应

接下来实现核心的响应生成系统与网页搜索能力，为研究自动化提供支撑。这里我们使用 Serper API，如下所示：

def generate_response(self, prompt: str, max_tokens: int = None, extra_stops: Optional[List[str]] = None) -> str:
        if not self.llm:
            return "Model not loaded."
        stops = ["</s>", "<|im_end|>", "<|endoftext|>"]
        if extra_stops:
            stops.extend(extra_stops)
        mt = max_tokens or self.config.max_tokens
        try:
            resp = self.llm(
                prompt,
                max_tokens=mt,
                temperature=self.config.temperature,
                top_p=self.config.top_p,
                top_k=self.config.top_k,
                stop=stops,
                echo=False
            )
            return resp["choices"][0]["text"].strip()
        except Exception as e:
            return f"Error generating response: {str(e)}"
    async def search_web(self, query: str, num_results: int = 10) -> List[Dict]:
        if self.config.search_api_key:
            return await self.search_serper(query, num_results)
        return []
    async def search_serper(self, query: str, num_results: int) -> List[Dict]:
        url = "https://google.serper.dev/search"
        payload = {"q": query, "num": num_results}
        headers = {"X-API-KEY": self.config.search_api_key, "Content-Type": "application/json"}
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload, headers=headers) as response:
                data = await response.json()
                results = []
                for item in data.get('organic', []):
                    results.append({
                        'title': item.get('title', ''),
                        'url': item.get('link', ''),
                        'snippet': item.get('snippet', ''),
                        'source': 'web'
                    })
                return results
    def generate_search_queries(self, topic: str, focus_area: str, depth: str) -> List[str]:
        counts = {'surface': 5, 'moderate': 8, 'deep': 15, 'comprehensive': 25}
        n = counts.get(depth, 8)
        base = [
            f"{topic} overview",
            f"{topic} recent developments",
            f"{topic} academic studies",
            f"{topic} case studies",
            f"{topic} policy and regulation",
            f"{topic} technical approaches",
            f"{topic} market analysis",
            f"{topic} statistics and data",
        ]
        return base[:n]技术支持

上述代码的核心功能拆解如下：

• generate_response() 函数：负责与 Jan-v1 的交互，具备完善的错误处理与可配置的停止序列，确保研究综合文本生成的一致性与高质量。
• search_web() 与 search_serper() 函数：使用 Serper API 实现异步网页搜索，支持对多条查询的快速并行检索；异步设计避免在研究过程中阻塞 UI。
• generate_search_queries() 函数：根据研究深度与关注主题生成多样化搜索查询，实现从综述、技术路径到案例、监管与市场分析的全覆盖。可按所选深度（surface、moderate、deep、comprehensive）将查询数在 5 至 25 条之间伸缩。

该响应生成系统是研究自动化的核心，将 Jan-v1 的推理能力与实时网页搜索结合，用于生成研究报告。

步骤 3.4：综合研究结果

接下来实现研究综合引擎，将原始搜索结果转化为打磨完备、专业规范的报告。

def synthesize_research(self, topic: str, search_results: List[Dict], focus_area: str, report_format: str) -> str:
        context_lines = []
        for i, result in enumerate(search_results[:20]):
            title = result.get("title", "")
            snippet = result.get("snippet", "")
            context_lines.append(f"Source {i+1} Title: {title}\nSource {i+1} Summary: {snippet}")
        context = "\n".join(context_lines)
        sections = sections_for_format(report_format)
        sections_text = "\n".join(sections)
        synthesis_prompt = f"""
You are an expert research analyst. Write the final, polished report on: "{topic}" for a professional, real-world audience.
***CRITICAL INSTRUCTIONS:***
- Your entire response MUST be the final report, wrapped **EXACTLY** inside <final> and </final> tags.
- DO NOT output any text, thoughts, or commentary BEFORE the <final> tag or AFTER the </final> tag.
- DO NOT include any conversational filler, internal thoughts, or commentary about the generation process (e.g., "As an AI...", "I will now summarize...", "Here is your report:").
- DO NOT use markdown formatting (e.g., #, ##, *, -).
- DO NOT use bullet points or lists.
- Maintain a formal, academic/professional tone throughout.
- Ensure the report is complete and self-contained.
- Include the following section headers, in this order, and no others:
{sections_text}
Guidance:
- Base your writing strictly on the Research Notes provided below. If the notes lack specific data, write a careful, methodology-forward analysis without inventing facts or numbers.
Research Notes:
{context}
Now produce ONLY the final report:
<final>
...your report here...
</final>
"""
        raw = self.generate_response(synthesis_prompt, max_tokens=1800, extra_stops=["</final>"])
        final_report = extract_final_block(raw)
        final_report = re.sub(r"(?m)^\s*[-*•]\s+", "", final_report)
        final_report = re.sub(r"[#`*_]{1,3}", "", final_report)
        first = next((h for h in sections if h in final_report), None)
        if first:
            final_report = final_report[final_report.find(first):].strip()
        return final_report技术支持

上述代码利用 Jan-v1 的高级推理能力，将原始网页搜索结果转化为通常需要数小时手工研究与写作才能完成的专业报告。其关键能力包括：

• 上下文嵌入： 将非结构化的搜索结果转化为结构化文本化的研究笔记，通过按序号拼接标题与摘要，提升信息密度、减少传入模型时的上下文碎片化。
• 动态提示构建： 借助 sections_for_format() 依据所选输出格式生成章节标题，并以严格规则注入提示，约束模型行为。
• 降低幻觉： 通过强制使用 <final>...</final> 标签并禁止对话式脚手架，避免模型引入无关文本、臆测性总结或重复性框架。
• 后处理与归一化： 解码后利用 regex 过滤器去除 Markdown 痕迹、列表标记与零散格式，并确保输出从首个有效章节标题开始，以增强结构一致性。

步骤 4：主 Streamlit 应用流程

最后我们构建完整的用户界面，将上述研究能力集成为一个交互式应用。

def main():
    st.set_page_config(page_title="Deep Research Assistant", page_icon="", layout="wide")
    st.markdown("""
    <div style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); padding: 1.25rem; border-radius: 10px; color: white; text-align: center; margin-bottom: 1rem;">
        <h1 style="margin:0;"> Deep Research Assistant</h1>
    </div>
    """, unsafe_allow_html=True)
    if 'research_assistant' not in st.session_state:
        config = ResearchConfig()
        st.session_state.research_assistant = DeepResearchAssistant(config)
        st.session_state.model_loaded = st.session_state.research_assistant.load_model()
        st.session_state.research_results = None
    if not st.session_state.research_assistant.config.search_api_key:
        st.warning("SERPER_API_KEY not found in environment. Using fallback demo results.")
    st.header("Research Configuration")
    research_topic = st.text_area(
        "Research Topic",
        placeholder="e.g., Impact of artificial intelligence on healthcare efficiency and patient outcomes",
        height=100
    )
    col1a, col1b = st.columns(2)
    with col1a:
        research_depth = st.selectbox(
            "Research Depth",
            ["surface", "moderate", "deep", "comprehensive"],
            index=1,
            format_func=lambda x: {
                "surface": "Surface (5-8 sources)",
                "moderate": "Moderate (10-15)",
                "deep": "Deep Dive (20-30)",
                "comprehensive": "Comprehensive (40+)"
            }[x]
        )
        focus_area = st.selectbox("Focus Area", ["general", "academic", "business", "technical", "policy"], index=0, format_func=str.title)
    with col1b:
        time_frame = st.selectbox(
            "Time Frame",
            ["current", "recent", "comprehensive"],
            index=1,
            format_func=lambda x: {"current":"Current (≤6 months)","recent":"Recent (≤2 years)","comprehensive":"All time"}[x]
        )
        report_format = st.selectbox(
            "Report Format",
            ["executive", "detailed", "academic", "presentation"],
            index=1,
            format_func=lambda x: {
                "executive":"Executive Summary",
                "detailed":"Detailed Analysis",
                "academic":"Academic Style",
                "presentation":"Presentation Format"
            }[x]
        )
    if st.button("Start Deep Research", type="primary", use_container_width=True):
        if not st.session_state.model_loaded:
            st.error("Model not loaded. See terminal logs for details.")
        elif not research_topic.strip():
            st.error("Please enter a research topic.")
        else:
            start_research(research_topic, research_depth, focus_area, time_frame, report_format)
    if st.session_state.research_results:
        display_research_results(st.session_state.research_results)
def start_research(topic: str, depth: str, focus: str, timeframe: str, format_type: str):
    assistant = st.session_state.research_assistant
    progress_bar = st.progress(0)
    status_text = st.empty()
    try:
        status_text.text("Generating search queries...")
        progress_bar.progress(15)
        queries = assistant.generate_search_queries(topic, focus, depth)
        status_text.text("Searching sources...")
        progress_bar.progress(40)
        all_results = []
        for i, query in enumerate(queries):
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            results = loop.run_until_complete(assistant.search_web(query, 5))
            all_results.extend(results)
            loop.close()
            progress_bar.progress(40 + int(((i + 1) / max(1, len(queries))) * 30))
            time.sleep(0.05)
        status_text.text("Synthesizing report...")
        progress_bar.progress(80)
        research_report = assistant.synthesize_research(topic, all_results, focus, format_type)
        status_text.text("Done.")
        progress_bar.progress(100)
        st.session_state.research_results = {
            'topic': topic,
            'report': research_report,
            'sources': all_results,
            'queries': queries,
            'config': {'depth': depth, 'focus': focus, 'timeframe': timeframe, 'format': format_type},
            'timestamp': datetime.now()
        }
        time.sleep(0.3)
        status_text.empty()
        progress_bar.empty()
    except Exception as e:
        st.error(f"Research failed: {str(e)}")
        status_text.empty()
        progress_bar.empty()
def display_research_results(results: Dict):
    st.header("Research Report")
    st.subheader(f"Topic: {results['topic']}")
    st.markdown(
        f'<div style="background:#f8f9ff;padding:1rem;border-radius:10px;border:1px solid #e1e8ed;">{results["report"]}</div>',
        unsafe_allow_html=True,
    )
    with st.expander("Sources", expanded=False):
        for i, source in enumerate(results['sources'][:12]):
            st.markdown(f"""
            <div style="background:#fff;padding:0.75rem;border-radius:8px;border:1px solid #e1e8ed;margin:0.4rem 0;">
                <h4 style="margin:0 0 .25rem 0;">{source['title']}</h4>
                <p style="margin:0 0 .25rem 0;">{source['snippet']}</p>
                <small><a href="{source['url']}" target="_blank">{source['url']}</a></small>
            </div>
            """, unsafe_allow_html=True)
    st.markdown("### Export")
    c1, c2, c3 = st.columns(3)
    with c1:
        report_text = f"Research Report: {results['topic']}\n\n{results['report']}"
        st.download_button("Download Text", data=report_text, file_name=f"research_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", mime="text/plain")
    with c2:
        json_data = json.dumps(results, default=str, indent=2)
        st.download_button("Download JSON", data=json_data, file_name=f"research_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", mime="application/json")
    with c3:
        if st.button("Start New Research"):
            st.session_state.research_results = None
            st.experimental_rerun()
if __name__ == "__main__":
    main()
技术支持

上述代码自下而上划分为多个层次：从模型加载与用户输入采集，到查询生成、网页搜索，再到最终报告综合。各层次要点如下：

• UI 层（Streamlit 前端）： 应用提供渐变风格的页眉与响应式布局，带来精致的用户体验；界面直观，同时提供细粒度研究参数控制，适合技术与非技术用户。
• 可配置的流水线： 用户可通过选择器自定义研究深度、关注领域（general、academic、business 等）、时间范围与输出风格（executive、academic 等）。
• 有状态的代理管理：DeepResearchAssistant 通过 st.session_state 在会话中只初始化一次，保留模型状态，避免重复加载。
• 实时反馈与异步搜索：借助 Streamlit 动态进度条与状态更新进行进度跟踪。网页搜索通过 asyncio 异步执行，在并行查询的同时保持界面响应。
• 查询与综合循环：基于主题与深度自动生成搜索查询（提示工程策略），聚合检索结果后交由 Jan-v1 生成结构化报告。
• 输出可视化与导出：最终报告以 Markdown 区块呈现，并配有可展开的引用来源。用户可将报告导出为 TXT 或 JSON，或一键重新运行流程。

若要亲自尝试，将代码保存为 app.py 并启动：

streamlit run app.py技术支持