这个项目目前最大的风险是什么？

上周提到的阻塞点解决了吗？

合同里关于交付延期是怎么约定的？

用户问题 → 大模型直接回答

用户问题 → 检索相关文档片段 → 把片段交给模型 → 基于片段回答

请根据 test_docs 里的资料，回答这个项目目前有哪些风险？

回答项目当前主要风险包括交付延期、接口依赖未确认、测试资源不足。依据1. sample_report.pdf / 第 2 页，提到接口联调依赖外部系统确认。2. meeting_notes.md，提到测试环境还未准备完成。

07_rag_qa.py

from dataclasses import dataclass@dataclassclass DocumentChunk: content: str source: str index: int

from pathlib import Pathfrom docx import Documentfrom pypdf import PdfReaderdef read_pdf(file_path: str) -> str: reader = PdfReader(file_path) pages = [] for i, page in enumerate(reader.pages): text = page.extract_text() or ”” text = text.strip() if text: pages.append(f”[第 {i + 1} 页]\n{text}”) return ”\n\n”.join(pages)def read_docx(file_path: str) -> str: doc = Document(file_path) parts = [] for p in doc.paragraphs: text = p.text.strip() if text: parts.append(text) return ”\n”.join(parts)def read_txt(file_path: str) -> str: return Path(file_path).read_text(encoding=”utf-8”)def read_document(file_path: str) -> str: path = Path(file_path) suffix = path.suffix.lower() if suffix == ”.pdf”: return read_pdf(str(path)) if suffix == ”.docx”: return read_docx(str(path)) if suffix in [”.txt”, ”.md”, ”.log”]: return read_txt(str(path)) return ””

def load_documents(folder: str) -> dict[str, str]: folder_path = Path(folder) docs = {} for file_path in folder_path.rglob(”*”): if not file_path.is_file(): continue if file_path.suffix.lower() not in [”.pdf”, ”.docx”, ”.txt”, ”.md”, ”.log”]: continue text = read_document(str(file_path)) if text.strip(): docs[str(file_path)] = text return docs

def chunk_text(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]: text = text.strip() if not text: return [] chunks = [] start = 0 while start < len(text): end = start + chunk_size chunk = text[start:end].strip() if chunk: chunks.append(chunk) start = end - overlap if start < 0: start = 0 if start >= len(text): break return chunksdef build_chunks(folder: str) -> list[DocumentChunk]: docs = load_documents(folder) all_chunks = [] for source, text in docs.items(): chunks = chunk_text(text) for i, chunk in enumerate(chunks, start=1): all_chunks.append( DocumentChunk( content=chunk, source=source, index=i, ) ) return all_chunks

chunk_size = 800overlap = 100

import redef tokenize(text: str) -> list[str]: tokens = re.findall(r”[\u4e00-\u9fa5A-Za-z0-9]+”, text.lower()) return [t for t in tokens if len(t.strip()) > 1]def score_chunk(question: str, chunk: DocumentChunk) -> int: question_tokens = set(tokenize(question)) content = chunk.content.lower() score = 0 for token in question_tokens: if token in content: score += 1 return scoredef retrieve(question: str, chunks: list[DocumentChunk], top_k: int = 4) -> list[DocumentChunk]: scored = [] for chunk in chunks: score = score_chunk(question, chunk) if score > 0: scored.append((score, chunk)) scored.sort(key=lambda x: x[0], reverse=True) return [chunk for _, chunk in scored[:top_k]]

问题 → 找相关片段 → 返回 top_k 个片段

import osfrom langchain_openai import ChatOpenAIdef build_llm() -> ChatOpenAI: api_key = os.getenv(”DEEPSEEK_API_KEY”) if not api_key: raise RuntimeError(”请先设置环境变量 DEEPSEEK_API_KEY”) return ChatOpenAI( model=”deepseek-chat”, base_url=”https://api.deepseek.com/v1”, api_key=api_key, temperature=0, )

def answer_question(llm: ChatOpenAI, question: str, chunks: list[DocumentChunk]) -> str: if not chunks: return ”没有检索到相关内容，无法基于文档回答。” context_parts = [] for i, chunk in enumerate(chunks, start=1): context_parts.append( f”[资料 {i}]\n来源：{chunk.source}\n片段编号：{chunk.index}\n内容：\n{chunk.content}” ) context = ”\n\n”.join(context_parts) prompt = f”””你是一个文档问答助手。请只根据下面提供的资料回答问题。要求：- 不要编造资料中没有的信息- 如果资料不足，直接说“资料中未明确提到”- 回答要简洁- 最后列出你参考了哪些资料来源用户问题：{question}可参考资料：{context}请按下面格式输出：回答直接回答用户问题。# 依据列出引用的资料来源和片段编号。””” result = llm.invoke(prompt) return result.content

def rag_answer(folder: str, question: str) -> str: print(”正在读取并切分文档...”) chunks = build_chunks(folder) print(f”共生成 {len(chunks)} 个文档片段。”) print(”正在检索相关片段...”) related_chunks = retrieve(question, chunks, top_k=4) if not related_chunks: return ”没有检索到相关内容。你可以换一种问法，或者检查文档是否包含相关信息。” print(”检索到的片段：”) for chunk in related_chunks: print(f”- {chunk.source} / 片段 {chunk.index}”) print(”正在生成回答...”) llm = build_llm() return answer_question(llm, question, related_chunks)if __name__ == ”__main__”: folder = ”test_docs” question = ”这个项目目前有哪些风险？” answer = rag_answer(folder, question) print(”\n” + ”=” * 80) print(answer)

python 07_rag_qa.py

正在读取并切分文档...共生成 12 个文档片段。正在检索相关片段...检索到的片段：- test_docs/sample_report.pdf / 片段 2- test_docs/meeting_notes.md / 片段 1正在生成回答...

交付有什么不确定性？

上线时间存在延期风险。

切文档 → 建片段 → 检索相关内容 → 基于内容回答

https://gitee.com/ai-gravity-field/guide-to-ai-agent-exploration