From eebc926c51f90986384e7b8e089f2f12939c396c Mon Sep 17 00:00:00 2001 From: Hwe1916 Date: Sat, 24 Oct 2026 00:45:27 +0800 Subject: [PATCH] feat(web): add demo frontends for the RAG agent Streamlit UI and a FastAPI endpoint so the agent can be demonstrated and called outside a notebook. --- app_web.py | 129 ++++++++++++++++++++++++++++++++++++++++++++++++ model/server.py | 98 ++++++++++++++++++++++++++++++++++++ server_demo.py | 110 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 337 insertions(+) create mode 100644 app_web.py create mode 100644 model/server.py create mode 100644 server_demo.py diff --git a/app_web.py b/app_web.py new file mode 100644 index 0000000..79afd56 --- /dev/null +++ b/app_web.py @@ -0,0 +1,129 @@ +"""Streamlit UI for inspecting sources and asking evidence-grounded TDoc questions.""" + +import json + +import streamlit as st + +from agent_core import ProposalAgent + +st.set_page_config(page_title="3GPP 提案阅读助手", page_icon="📡", layout="wide") + + +@st.cache_resource(show_spinner="正在建立本地提案检索索引……") +def load_agent(): + return ProposalAgent() + + +def show_sources(sources): + """Keep source excerpts visible independently of model loading or generation errors.""" + for warning in sources.get("warnings", []): + st.warning(warning) + if sources.get("documents"): + st.dataframe([ + {"提案": item["tdoc"], "标题": item.get("title", ""), "原文状态": item["status"]} + for item in sources["documents"] + ], hide_index=True, width="stretch") + for item in sources.get("evidence", []): + with st.expander(f"[{item['source_id']}] {item['tdoc']} · 段落 {item['paragraph_start']}–{item['paragraph_end']}"): + st.write(item["title"]) + st.caption(f"{item['meeting']} | {item['company']}") + st.text(item["text"]) + st.caption(item["file_path"]) + + +st.title("📡 3GPP 提案阅读助手") +st.caption("查找本地提案原文，查看依据，再生成带来源编号的回答。提案主张不等于已通过的标准。") + +with st.sidebar: + st.subheader("使用方式") + retrieval_only = st.checkbox("只查原文，不调用模型", value=True) + use_lora = st.checkbox("使用现有 LoRA 适配器", value=False, help="默认使用原模型。本轮少量实测中，现有适配器出现了引用缺失和无依据的流程描述；其整体效果尚未确认。") + max_tokens = st.slider("最长生成 token 数", 128, 1536, 640, step=128) + temperature = st.slider("采样温度", 0.0, 0.7, 0.0, step=0.1) + st.caption("设为 0 可减少采样变化；不能保证事实正确。首次生成会加载本地模型。") + st.divider() + st.subheader("示例问题") + examples = { + "XR 速率控制": "请根据 R2-2408133 原文，说明提案要解决的问题、主要主张及尚未明确的部分。", + "比较两篇 XR 提案": "请对比 R2-2408133 和 R2-2408134，分别说明目标和主要主张，再解释两篇提案是否解决同一个问题。", + "NTN 设计要求": "请根据 R2-2508703 原文，说明 NTN 对 TN/NTN 共同设计提出了哪些要求。区分提案主张与已经达成的结论。", + "按标题找资料": "XR rate control signaling", + } + for label, example in examples.items(): + if st.button(label, use_container_width=True): + st.session_state["question"] = example + if st.button("清除结果", use_container_width=True): + st.session_state.pop("last_run", None) + st.session_state.pop("last_error", None) + st.session_state.pop("last_sources", None) + st.caption("支持编号检索、标题关键词排序和 DOCX 原文读取。旧版 .doc 暂不读取；缺少原文时不生成内容分析。") + +question = st.text_area("问题或提案编号", key="question", height=130, placeholder="例如：对比 R2-2408133 和 R2-2408134 的目标与主张。") +extra = st.text_area("补充材料（选填，会标记为未经核实）", height=80) + +if st.button("检索原文" if retrieval_only else "检索并分析", type="primary"): + st.session_state.pop("last_run", None) + st.session_state.pop("last_error", None) + st.session_state.pop("last_sources", None) + if not question.strip(): + st.warning("请输入问题或提案编号。") + else: + status = st.status("正在处理……", expanded=True) + with st.container(): + source_area = st.container() + output = st.empty() + completed = False + try: + agent = load_agent() + for event in agent.run_stream( + question, extra_context=extra, max_new_tokens=max_tokens, + temperature=temperature, retrieval_only=retrieval_only, use_lora=use_lora, + ): + stage = event["stage"] + if stage in {"1. RETRIEVAL", "1. RETRIEVAL_DONE", "2. SYNTHESIS_START"}: + status.write(event["msg"]) + if stage == "1. RETRIEVAL_DONE": + st.session_state["last_sources"] = event["retrieval"] + with source_area: + show_sources(event["retrieval"]) + elif stage == "2. STREAMING": + output.markdown(event["full_text"] + " ▌") + elif stage == "4. COMPLETE": + st.session_state["last_run"] = {**event, "question": question, "use_lora": use_lora} + completed = True + elif stage == "ERROR": + st.session_state["last_error"] = event["msg"] + except Exception as exc: + st.session_state["last_error"] = f"初始化失败：{exc}" + output.empty() + status.update(label="处理完成" if completed else "处理失败", state="complete" if completed else "error", expanded=False) + if st.session_state.get("last_error"): + st.warning("生成未完成。上方原文仍可查看；不完整回答未保存为结果。") + +if st.session_state.get("last_error"): + st.error(st.session_state["last_error"]) + if st.session_state.get("last_sources"): + with st.expander("查看本次已找到的原文"): + show_sources(st.session_state["last_sources"]) + +if run := st.session_state.get("last_run"): + st.divider() + if run.get("mode") == "needs_review": + st.warning("以下为待核对草稿：来源编号或提案引用不完整。") + st.caption("对应问题：" + run.get("question", "")) + st.markdown(run["result"]) + with st.expander("查看所有来源及原文"): + show_sources(run["sources"]) + for warning in run.get("validation", {}).get("warnings", []): + st.warning(warning) + metrics = run.get("metrics", {}) + labels = {"retrieval_seconds": "检索", "first_text_seconds": "首段文本", "total_seconds": "总耗时"} + st.caption(" | ".join(f"{labels.get(name, name)}: {value:.2f} s" for name, value in metrics.items() if value is not None)) + st.caption("首段文本时间包含检索和首次模型加载，不等于纯模型首 token 延迟。") + report = run["result"] + "\n\n---\n\n来源记录：\n\n" + "\n\n".join( + f"[{item['source_id']}] {item['tdoc']} | {item['title']}\n" + f"{item['file_path']}（段落 {item['paragraph_start']}–{item['paragraph_end']}）\n{item['text']}" + for item in run["sources"].get("evidence", []) + ) + st.download_button("下载回答与来源 Markdown", report, "3gpp_analysis.md", "text/markdown") + st.download_button("下载完整运行记录 JSON", json.dumps(run, ensure_ascii=False, indent=2), "3gpp_run.json", "application/json") diff --git a/model/server.py b/model/server.py new file mode 100644 index 0000000..dc91c3c --- /dev/null +++ b/model/server.py @@ -0,0 +1,98 @@ +import os +import json +import time +import torch +from threading import Thread +from fastapi import FastAPI +from fastapi.responses import StreamingResponse +from pydantic import BaseModel +from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig, TextIteratorStreamer +from peft import PeftModel +import uvicorn + +# 1. 实例化 FastAPI 异步 Web 服务 +app = FastAPI(title="3GPP-LLM-Inference-Server") + +# 2. 本地基础模型与我们微调好的 LoRA 权重路径 +MODEL_DIR = r"C:\Users\ZoOoM\.cache\modelscope\hub\models\Qwen\Qwen2___5-3B-Instruct" +LORA_DIR = "output/qwen3b_ran2_qlora" + +print("正在以 4-bit 精度加载 3GPP 垂直微调模型，保护 8G 显存...") + +# 3. 初始化分词器（Tokenizer） +tokenizer = AutoTokenizer.from_pretrained(MODEL_DIR, trust_remote_code=True) +if tokenizer.pad_token is None: + tokenizer.pad_token = tokenizer.eos_token + +# 4. 配置刚才学到的 4-bit NF4 量化（将 6GB 权重压缩至 1.5GB） +bnb_config = BitsAndBytesConfig( + load_in_4bit=True, + bnb_4bit_quant_type="nf4", + bnb_4bit_use_double_quant=True, + bnb_4bit_compute_dtype=torch.bfloat16 +) + +# 5. 加载基座 + 挂载我们微调的 LoRA 适配器 +base_model = AutoModelForCausalLM.from_pretrained( + MODEL_DIR, + quantization_config=bnb_config, + device_map="auto", + attn_implementation="sdpa", + trust_remote_code=True +).eval() + +model = PeftModel.from_pretrained(base_model, LORA_DIR).eval() +print("✅ 模型与 LoRA 适配器加载完成，推理服务就绪！") + + +# ==================== 6. 定义 OpenAI 兼容的请求体规范 ==================== +class ChatCompletionRequest(BaseModel): + messages: list # 必填：对话历史 [{"role": "user", "content": "..."}] + temperature: float = 0.2 # 采样温度（默认 0.2，保证严谨） + max_tokens: int = 512 # 最大生成长度 + stream: bool = True # 是否开启打字机流式传输 +# ==================== 7. 实现核心流式接口 /v1/chat/completions ==================== +@app.post("/v1/chat/completions") +async def chat_completions(req: ChatCompletionRequest): + # 1. 使用分词器内置模板格式化整个对话历史 + prompt_text = tokenizer.apply_chat_template( + req.messages, + tokenize=False, + add_generation_prompt=True + ) + inputs = tokenizer([prompt_text], return_tensors="pt").to("cuda:0") + # 2. 创建异步流式管道 + streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True) + + gen_kwargs = dict( + **inputs, + streamer=streamer, + max_new_tokens=req.max_tokens, + temperature=req.temperature, + top_p=0.9, + do_sample=True if req.temperature > 0.0 else False, + eos_token_id=tokenizer.eos_token_id + ) + # 3. 启动独立工作线程执行计算（不阻塞 HTTP 主通道） + thread = Thread(target=model.generate, kwargs=gen_kwargs) + thread.start() + # 4. 定义 SSE 数据生成器 + def sse_event_generator(): + for token_text in streamer: + chunk_data = { + "choices": [{ + "delta": {"content": token_text}, + "finish_reason": None + }] + } + # 按照标准 SSE 协议输出：data: \n\n + yield f"data: {json.dumps(chunk_data, ensure_ascii=False)}\n\n" + + # 结束标记（与 OpenAI 规范完全一致） + yield "data: [DONE]\n\n" + # 5. 返回长连接流式响应 + return StreamingResponse(sse_event_generator(), media_type="text/event-stream") +# ==================== 8. 启动脚本入口 ==================== +if __name__ == "__main__": + # 在本地 8000 端口启动服务 + uvicorn.run(app, host="127.0.0.1", port=8000) \ No newline at end of file diff --git a/server_demo.py b/server_demo.py new file mode 100644 index 0000000..532a35a --- /dev/null +++ b/server_demo.py @@ -0,0 +1,110 @@ +"""Local HTTP entry points sharing the same lazy inference and evidence workflow.""" + +import json +import time +import uuid +from functools import lru_cache +from typing import Literal + +from fastapi import FastAPI, HTTPException +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field + +from agent_core import ProposalAgent + +app = FastAPI(title="3GPP Proposal Assistant") + + +@lru_cache(maxsize=1) +def get_agent(): + return ProposalAgent() + + +class Message(BaseModel): + role: Literal["system", "user", "assistant"] + content: str = Field(max_length=16000) + + +class ChatCompletionRequest(BaseModel): + model: str = "local-qwen" + messages: list[Message] = Field(min_length=1, max_length=32) + temperature: float = Field(default=0.0, ge=0, le=2) + max_tokens: int = Field(default=512, ge=1, le=2048) + stream: bool = True + use_lora: bool = False + + +class AnalysisRequest(BaseModel): + question: str = Field(min_length=1, max_length=8000) + extra_context: str = Field(default="", max_length=8000) + retrieval_only: bool = True + use_lora: bool = False + max_tokens: int = Field(default=640, ge=1, le=2048) + + +def encode_sse(data): + return f"data: {json.dumps(data, ensure_ascii=False)}\n\n" + + +@app.get("/health") +def health(): + return {"status": "ok", "model_loaded": bool(get_agent().runtime.model is not None)} + + +@app.post("/v1/proposals/analyze") +def analyze(req: AnalysisRequest): + """SSE workflow including sources, explicit failures and retrieval-only operation.""" + def events(): + for event in get_agent().run_stream( + req.question, extra_context=req.extra_context, retrieval_only=req.retrieval_only, + use_lora=req.use_lora, max_new_tokens=req.max_tokens, + ): + yield encode_sse(event) + yield "data: [DONE]\n\n" + return StreamingResponse(events(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"}) + + +@app.post("/v1/chat/completions") +def chat_completions(req: ChatCompletionRequest): + """Basic chat-compatible endpoint; raw chat does not run source retrieval.""" + completion_id = "chatcmpl-" + uuid.uuid4().hex + created = int(time.time()) + + def response_base(kind): + return {"id": completion_id, "object": kind, "created": created, "model": req.model} + + chunks = get_agent().runtime.stream( + [message.model_dump() for message in req.messages], req.max_tokens, req.temperature, req.use_lora, + ) + if not req.stream: + try: + text = "".join(chunks) + except Exception as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + return {**response_base("chat.completion"), "choices": [ + {"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"} + ]} + + def events(): + try: + yield encode_sse({**response_base("chat.completion.chunk"), "choices": [ + {"index": 0, "delta": {"role": "assistant"}, "finish_reason": None} + ]}) + for chunk in chunks: + yield encode_sse({**response_base("chat.completion.chunk"), "choices": [ + {"index": 0, "delta": {"content": chunk}, "finish_reason": None} + ]}) + yield encode_sse({**response_base("chat.completion.chunk"), "choices": [ + {"index": 0, "delta": {}, "finish_reason": "stop"} + ]}) + except Exception as exc: + yield encode_sse({"error": {"message": str(exc), "type": type(exc).__name__}}) + finally: + chunks.close() + yield "data: [DONE]\n\n" + return StreamingResponse(events(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"}) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="127.0.0.1", port=8000)