import asyncio import re import sys from typing import Optional, List, Set,Iterable if sys.platform.startswith("win") and hasattr(asyncio, "WindowsSelectorEventLoopPolicy"): asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) try: import ahocorasick except ImportError: ahocorasick = None from pydantic import BaseModel, Field, model_validator from fastapi import FastAPI, File, Path, UploadFile, HTTPException, Form, Request, Header, Body, BackgroundTasks from fastapi.responses import StreamingResponse try: from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate except ImportError: try: from langchain.prompts import FewShotPromptTemplate, PromptTemplate except ImportError: class PromptTemplate: def __init__(self, input_variables=None, template: str = ""): self.input_variables = input_variables or [] self.template = template def format(self, **kwargs): return self.template.format(**kwargs) class FewShotPromptTemplate: def __init__( self, examples=None, example_prompt=None, prefix: str = "", suffix: str = "", input_variables=None, example_separator: str = "\n\n", ): self.examples = examples or [] self.example_prompt = example_prompt self.prefix = prefix self.suffix = suffix self.input_variables = input_variables or [] self.example_separator = example_separator def format(self, **kwargs): rendered_examples = [] for example in self.examples: rendered_examples.append(self.example_prompt.format(**example)) parts = [self.prefix] if rendered_examples: parts.append(self.example_separator.join(rendered_examples)) parts.append(self.suffix.format(**kwargs)) return self.example_separator.join(part for part in parts if part) from openai.types.chat import ChatCompletionSystemMessageParam, ChatCompletionUserMessageParam, ChatCompletionMessageParam from chunk_text import ( get_chunk_bbox, split_text_preserve_sentences, data_replace, chunk_check, merge_short_slices, find_and_read_content_list, process_pdf_file, process_other_file, safe_original_filename, image_base64_to_data_url, extract_image_text, prepare_split_image, ) from fastapi.responses import JSONResponse, StreamingResponse import requests from pathlib import PurePath, Path as PathLib import tempfile from modelsAPI import model_api from documents_prompt import ( BEAUTIFY_TYPE_DICT, TEMPLATE_DICT, GENERATE_NORMAL_OUTLINE_PROMPT, GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT, GENERATE_METHOD_OUTLINE_PROMPT, ) """app.py FastAPI 主 Agent 接口 - 修复重复执行问题 提供主脑 Agent 的 HTTP API 服务，支持流式输出 集成 LangGraph Checkpointer 实现状态持久化 """ import logging import unicodedata from time import sleep import aiofiles from dotenv import load_dotenv from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse, FileResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import Optional, Dict, Any, AsyncGenerator,List import uvicorn import json import os import uuid from main_agent import create_main_agent from utils.function_tracker import ( register_event_callback, create_stream_event_handler, unregister_event_callback, clear_current_stream_handler ) import asyncio import importlib from doc2pdf import Doc2PDF from workflow_registry import WORKFLOW_CONFIG, VALID_ROUTE_FLAGS from main_agent import extract_conversation_title from workflows.history_manager import filter_image_urls, preprocess_from_request from config import set_request_user_config from checkpointer_config import ( CheckpointerManager, checkpointer_manager ) from gw_write import cotprompt,build_review_prompt,length_convert,length_display,build_length_instruction,dedupe_and_filter,filter_positive_word_false_positives,append_dictionary_results,find_keywords,normalize_words,process_rag_prompt app = FastAPI(max_request_size=1024 * 1024 * 10) _doc2pdf_converter: Optional[Doc2PDF] = None DATA_DIR = "/app/files" load_dotenv() VLM_SEMAPHORE = asyncio.Semaphore(1) logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) REWRITE_TYPE_MAP = { "quotePolish": "金句润色", "formalStyle": "书面化", "speechStyle": "讲稿化", "formatting": "规整", "expand": "扩写", "continueWriting": "续写", "report": "汇报", "quote": "金句", "summarize": "精简", "condense": "总结", } # /Review 流式并发配置：块大小 5000，最多 5 块并行 REVIEW_CHUNK_SIZE = 1000 REVIEW_MAX_CONCURRENCY = 5 REVIEW_SYSTEM_PROMPT = ( "你是一个严谨、穷尽式的专业文本校对专家，必须完整检查全文并返回所有符合规则的问题，不能只挑选部分问题。" "校对结果会由程序直接替换进正式公文，因此建议字段只能包含最终正文，任何解释、括号备注或操作说明都会污染公文，绝对禁止输出。" "每条原文字段必须是输入正文中真实存在的连续原样子串，不能包含任何提前修改后的文字，否则该结果会被系统丢弃。" "请严格按照用户要求的格式输出。" "不要输出推理过程、解释性文字、前缀或总结。" ) def get_doc2pdf_converter() -> Doc2PDF: global _doc2pdf_converter if _doc2pdf_converter is None: try: _doc2pdf_converter = Doc2PDF() except RuntimeError as exc: logger.error(f"LibreOffice 初始化失败: {exc}") raise HTTPException(status_code=500, detail=str(exc)) from exc return _doc2pdf_converter app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # app.mount("/picture", StaticFiles(directory="/app/picture"), name="pictures") STREAM_RESPONSE_HEADERS = { "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "X-Content-Type-Options": "nosniff", } def llm_text_stream_response(content) -> StreamingResponse: return StreamingResponse( content, media_type="text/plain; charset=utf-8", headers=STREAM_RESPONSE_HEADERS, ) @app.on_event("startup") async def startup_event(): """应用启动时初始化""" await checkpointer_manager.setup() # 初始化智能体使用统计表 from tools.agent_usage_statistics import init_agent_usage_table await init_agent_usage_table() logger.info("应用启动完成，PostgreSQL Checkpointer 已配置") # =============== PDF 文件目录配置 =============== # 创建 PDF 文件目录（用于存储生成的 PDF 等文件） PDF_DIR = os.path.join(os.path.dirname(__file__), "created_pdf") # 创建 word 文件目录（用于存储生成的 word 等文件） WORD_DIR = os.path.join(os.path.dirname(__file__), "created_word") os.makedirs(PDF_DIR, exist_ok=True) class RewriteRequest(BaseModel): rewrite_type: Optional[str] = None content: Optional[str] = None require: Optional[str] = None sentence: Optional[str] = None def build_rewrite_prompt(request: RewriteRequest) -> str: beautify_type = REWRITE_TYPE_MAP.get(request.rewrite_type or "") if not beautify_type: raise HTTPException(status_code=400, detail=f"不支持的 rewrite_type: {request.rewrite_type}") prompt_template = BEAUTIFY_TYPE_DICT.get(beautify_type) if not prompt_template: raise HTTPException(status_code=400, detail=f"未找到提示词模板: {beautify_type}") return prompt_template.format( require=request.require or "", content=request.content or "", sentence=request.sentence or request.content or "", ) async def stream_llm_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt="你是专业文稿改写助手，根据要求进行文本改写。", messages=[], ): if chunk: yield chunk @app.post("/Rewrite") async def rewrite(request: RewriteRequest): prompt = build_rewrite_prompt(request) return llm_text_stream_response(stream_llm_content(prompt)) class WriteRequest(BaseModel): role: Optional[str] = None template: Optional[str] = None title: Optional[str] = None length: Optional[str] = None requirement: Optional[str] = None references: Optional[str] = None outline: Optional[str] = None konwlege: Optional[str] = None prompt: Optional[str] = None class OutlineRequest(BaseModel): role: Optional[str] = None template: Optional[str] = None title: Optional[str] = None length: Optional[str] = None requirement: Optional[str] = None references: Optional[str] = None konwlege: Optional[str] = None prompt: Optional[str] = None WRITE_FORMULA_INSTRUCTION = r""" 【公式输出规范（必须遵守）】 1. 仅在正文确实需要数学、统计或技术公式时使用 LaTeX；普通数字、百分比、日期、编号和金额均使用普通文本，不要为了排版而生成公式。 2. 行内公式必须且只能写成 `$公式内容$`。开始和结束的 `$` 之间不得换行。 3. 独占一行的公式必须使用成对的 `$$`，公式内容放在两者之间。 4. 只输出 KaTeX 支持的标准 LaTeX。禁止使用 `\(...\)`、`\[...\]`、LaTeX 文档环境、代码块、HTML、MathML 或 JSON 包裹公式。 5. 每个公式的定界符必须成对闭合。公式中的说明文字使用 `\text{...}`；百分号写作 `\%`，乘号优先写作 `\times`。 6. 金额直接写成“100元”“20万元”等普通文本，不得使用 `$` 作为货币符号。 7. 若正文不需要公式，不要输出任何 `$` 或 `$$`。 """ WRITE_SYSTEM_PROMPT = ( "你是专业公文写作助手，请直接输出正文内容。" "输出使用可由前端 Markdown 和 KaTeX 直接解析的内容；如需公式，必须严格遵守用户提示中的公式输出规范。" ) FIRST_LEVEL_HEADING_PATTERN = re.compile( r"(?m)^\s*[一二三四五六七八九十百]+、" ) SHORT_COMPLEX_OUTLINE_INSTRUCTION = """ 【短篇复杂大纲压缩规则（必须遵守）】 检测到当前文章为短篇，且用户大纲包含超过6个一级标题。为确保全文不超过1000字，必须遵守以下规则： 1. 一级标题最多保留6个。大纲超过6个一级标题时，必须合并内容相近的部分，不得逐项机械展开。 2. 合并大纲时不得删除核心事项，但允许调整标题名称及层级；本规则优先级高于“严格保持原大纲格式、标题不变”的要求。 3. 开头导语控制在120字以内。 4. 每个一级标题的全部内容控制在100字以内。 5. 每个一级标题下最多保留2个二级事项；超过2个时必须合并同类内容。 6. 每个二级事项只使用一句话表述，控制在40字以内，不得继续设置三级标题。 7. 背景、目的和意义应合并表述；纪律要求、成果转化和考核要求应合并表述。 8. 师资介绍、后勤保障、附件说明等次要内容应简写，不得展开评价性、宣传性描述。 9. 参考材料只用于提取时间、地点、对象、内容、要求等关键信息，不得逐段复述。 10. 全文目标约700字，绝对不得超过1000字；篇幅上限优先于大纲层级和参考材料完整复述。 """ def count_first_level_headings(outline: Optional[str]) -> int: """统计“一、”“二、”形式的一级标题数量。""" return len(FIRST_LEVEL_HEADING_PATTERN.findall(outline or "")) def build_write_prompt(request: WriteRequest) -> str: prompt = request.prompt or "" template = request.template or "" length_instruction = build_length_instruction(request.length) content = f"请撰写一篇{template}。{length_instruction}{prompt}" prompt_template = TEMPLATE_DICT.get(template) if not prompt_template: raise HTTPException(status_code=400, detail=f"不支持的写作模板: {template}") title = request.title or "" if not title.strip(): raise HTTPException(status_code=400, detail="title不能为空") document_prompt = content + prompt_template.format( role=request.role or "", title=title, length=length_display(request.length), requirement=request.requirement or "", references=request.references or "", outline=request.outline or "", ) structure_instruction = "" first_level_count = count_first_level_headings(request.outline) if request.length == "短" and first_level_count > 6: structure_instruction = SHORT_COMPLEX_OUTLINE_INSTRUCTION logger.info( "[公文撰写] 短篇大纲包含%s个一级标题，启用复杂大纲压缩规则", first_level_count, ) return document_prompt + structure_instruction + WRITE_FORMULA_INSTRUCTION async def stream_write_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt=WRITE_SYSTEM_PROMPT, messages=[], ): if chunk: yield chunk @app.post("/Write") async def write(request: WriteRequest): prompt = build_write_prompt(request) return llm_text_stream_response(stream_write_content(prompt)) def build_outline_prompt(request: OutlineRequest) -> str: if request.prompt and request.prompt.strip(): return request.prompt.strip() template = request.template or "" title = request.title or "" if not title.strip(): raise HTTPException(status_code=400, detail="title不能为空") markdown_list_template = {"报告", "函", "文书", "工作总结"} prompt_template = GENERATE_NORMAL_OUTLINE_PROMPT if template == "方案": prompt_template = GENERATE_METHOD_OUTLINE_PROMPT elif template in markdown_list_template: prompt_template = GENERATE_MARKDOWN_LIST_OUTLINE_PROMPT return prompt_template.format( role=request.role or "", template=template, title=title, length=length_display(request.length), requirement=request.requirement or "", references=request.references or "", ) async def stream_outline_content(prompt: str): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( query=prompt, model=None, system_prompt="你是专业公文提纲生成助手，请直接输出提纲内容。", messages=[], ): if chunk: yield chunk @app.post("/Outline") async def outline(request: OutlineRequest): prompt = build_outline_prompt(request) return llm_text_stream_response(stream_outline_content(prompt)) class ReviewRequest(BaseModel): require: Optional[str] = Field(None, description="额外校对要求") content: str = Field(..., description="待校对文本") types: List[str] = Field(..., description="检查类型") sensitive_words: List[str] = Field(default_factory=list, description="敏感词汇") negative_words: List[str] = Field(default_factory=list, description="错误词汇") positive_words: List[str] = Field(default_factory=list, description="正词词汇") @model_validator(mode="after") def check_content(self): if not self.content or not self.content.strip(): raise ValueError("content不能为空") return self class ReviewItem(BaseModel): original: str error: str suggestion: str class ReviewResponse(BaseModel): data: List[ReviewItem] _REVIEW_BLOCK_RE = re.compile( r"错误[:：]\s*(?P.*?)\s*[\r\n]+" r"原文[:：]\s*(?P.*?)\s*[\r\n]+" r"建议[:：]\s*(?P.*?)(?=\n\s*\n|$)", re.S, ) def parse_review_output(text: str) -> List[dict]: text = text.strip() if text.startswith("###"): text = text[3:].strip() if text.endswith("###"): text = text[:-3].strip() result = [] for match in _REVIEW_BLOCK_RE.finditer(text): item = { "error": match.group("error").strip(), "original": match.group("original").strip(), "suggestion": match.group("suggestion").strip(), } # 原文字段为空或仅包含空白时，不作为有效校对记录。 if item["original"]: result.append(item) return result def _split_review_error_types(error: str) -> List[str]: """按原顺序拆分并去重校对错误类型。""" result = [] for error_type in re.split(r"[、,，]", error or ""): error_type = error_type.strip() if error_type and error_type not in result: result.append(error_type) return result def _is_whitespace_only_change(original: str, suggestion: str) -> bool: """判断原文与建议是否仅存在空白或末尾标点差异。""" if not original or not suggestion or original == suggestion: return False def normalize(value: str) -> str: # 统一全角/半角字符，避免“(4)”与“（4）”被误判为内容修改。 value = unicodedata.normalize("NFKC", str(value)) value = re.sub(r"\s+", "", value) # 校对模型有时会给原文补充句末冒号、句号等格式标点； # 若除此之外内容完全一致，则不作为实际错误返回。 return re.sub(r"[，。；：、,.!?！？:;]+$", "", value) return normalize(original) == normalize(suggestion) def repair_review_output( model_output: str, source_text: str, positive_words: Optional[Set[str]] = None, ) -> str: """修复提前改写原文以及修改范围重叠的校对结果。""" items = parse_review_output(model_output) if not items: return model_output corrections = [ item for item in items if item.get("original") and item.get("suggestion") and item["original"] != item["suggestion"] and not _is_whitespace_only_change(item["original"], item["suggestion"]) ] corrections.sort(key=lambda item: len(item["suggestion"]), reverse=True) repaired_items = [] for item in items: repaired = dict(item) if ( repaired.get("original", "").strip() == "核心母" and repaired.get("suggestion", "").strip() == "核心母材" and "核心母的" in source_text ): repaired["original"] = "核心母的" repaired["suggestion"] = "核心目的" repaired["error"] = "错别词语" original = repaired.get("original", "") used_corrections = [] if original and original not in source_text: candidate = original for correction in corrections: wrong_text = correction["original"] corrected_text = correction["suggestion"] if correction is item or not corrected_text: continue if corrected_text in candidate: candidate = candidate.replace(corrected_text, wrong_text) used_corrections.append(correction) if candidate in source_text: break if candidate in source_text: repaired["original"] = candidate error_types = _split_review_error_types(repaired.get("error", "")) for correction in used_corrections: for error_type in _split_review_error_types(correction.get("error", "")): if error_type not in error_types: error_types.append(error_type) repaired["error"] = "、".join(error_types) repaired_items.append(repaired) covered_indexes = set() for long_index, long_item in enumerate(repaired_items): long_original = long_item.get("original", "") long_suggestion = long_item.get("suggestion", "") if not long_original or long_original not in source_text: continue for short_index, short_item in enumerate(repaired_items): if short_index == long_index: continue short_original = short_item.get("original", "") short_suggestion = short_item.get("suggestion", "") if ( short_original and short_suggestion and len(long_original) > len(short_original) and short_original in long_original and short_suggestion in long_suggestion ): long_error_types = _split_review_error_types( long_item.get("error", "") ) for error_type in _split_review_error_types( short_item.get("error", "") ): if error_type not in long_error_types: long_error_types.append(error_type) long_item["error"] = "、".join(long_error_types) covered_indexes.add(short_index) final_items = [ item for index, item in enumerate(repaired_items) if index not in covered_indexes and item.get("original", "").strip() and item.get("error", "").strip() and not _is_whitespace_only_change( item.get("original", ""), item.get("suggestion", "") ) ] # 正词命中的“错别字”记录不视为错误，例如“维休”被定义为正词时， # 不返回“维休”→“维修”的错别字建议。 final_items = filter_positive_word_false_positives( final_items, positive_words or set(), ) blocks = [ "\n".join([ f"错误：{item.get('error', '').strip()}", f"原文：{item.get('original', '').strip()}", f"建议：{item.get('suggestion', '').strip()}", ]) for item in final_items if item.get("original", "").strip() ] return "###\n" + "\n\n".join(blocks) + "\n###" async def stream_review_model(prompt: Iterable[ChatCompletionMessageParam]): async for chunk in model_api.OpenaiAPI.open_api_chat_stream( model=None, json_output=False, system_prompt=REVIEW_SYSTEM_PROMPT, messages=prompt, enable_thinking=False, temperature=0.1, ): if chunk: yield chunk async def call_review_model(prompt: Iterable[ChatCompletionMessageParam]) -> str: full_text = "" async for chunk in stream_review_model(prompt): full_text += chunk return full_text def build_review_stream_require( require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ) -> Optional[str]: requirements = [] if require and require.strip(): requirements.append(require.strip()) sensitive_text = "、".join(dict.fromkeys(word.strip() for word in sensitive_words if word and word.strip())) if sensitive_text: requirements.append(f"请重点检查以下敏感词：{sensitive_text}。如果原文出现这些词，错误类型写“敏感词汇”。") negative_text = "、".join(dict.fromkeys(word.strip() for word in negative_words if word and word.strip())) if negative_text: requirements.append(f"请重点检查以下错误词：{negative_text}。如果原文出现这些词，错误类型写“错误词汇”。") positive_text = "、".join(dict.fromkeys(word.strip() for word in positive_words if word and word.strip())) if positive_text: requirements.append(f"以下词为正确词，涉及错别字判断时不要误判：{positive_text}。") if not requirements: return None return "\n".join(requirements) def split_review_chunks(content: str, max_size: int = REVIEW_CHUNK_SIZE) -> List[str]: """优先按段落、其次按句末标点切分，单块不超过 max_size。""" if not content: return [] if max_size <= 0: raise ValueError("max_size must be greater than zero") chunks: List[str] = [] start = 0 while start < len(content): if len(content) - start <= max_size: chunks.append(content[start:]) break candidate = content[start:start + max_size] paragraph_matches = list(re.finditer(r"\n\s*", candidate)) if paragraph_matches: cut = paragraph_matches[-1].end() else: sentence_matches = list(re.finditer( r"[。！？；.!?;](?:[\"'”’）)】》])?\s*", candidate)) cut = sentence_matches[-1].end() if sentence_matches else max_size chunks.append(content[start:start + cut]) start += cut return chunks async def stream_review_content( content: str, types: List[str], require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ): review_require = build_review_stream_require( require=require, sensitive_words=sensitive_words, negative_words=negative_words, positive_words=positive_words, ) chunks = split_review_chunks(content, REVIEW_CHUNK_SIZE) if not chunks: return semaphore = asyncio.Semaphore(REVIEW_MAX_CONCURRENCY) async def process_chunk(chunk: str) -> str: async with semaphore: prompt = build_review_prompt( content=chunk, types=types, require=review_require, ) # 累积单块完整输出后再返回：并发块之间不能交错 token， # 否则会破坏“错误/原文/建议”块结构，导致后端解析错乱。 model_output = await call_review_model(prompt) repaired_output = repair_review_output( model_output, chunk, positive_words=normalize_words(positive_words), ) # 仅打印过滤后的最终保留记录，不打印模型原始输出或中间结果。 if repaired_output.strip() and repaired_output.strip() != "###\n###": logger.info("[公文校对] 最终保留记录：\n%s", repaired_output) return repaired_output tasks = [asyncio.create_task(process_chunk(chunk)) for chunk in chunks] # 谁先完成谁先产出（块间顺序不保证；校对条目相互独立，前端按原文定位） for finished in asyncio.as_completed(tasks): chunk_result = await finished if chunk_result and chunk_result.strip(): yield chunk_result yield "\n\n" # 块间分隔，避免相邻块内容被后端按 \n\n 粘连 async def review_chunk( content: str, types: List[str], require: Optional[str], ) -> List[dict]: prompt = build_review_prompt( content=content, types=types, require=require, ) model_output = await call_review_model(prompt) return parse_review_output(model_output) async def review_text( content: str, types: List[str], require: Optional[str], sensitive_words: List[str], negative_words: List[str], positive_words: List[str], ) -> List[dict]: sensitive_word_set = normalize_words(sensitive_words) negative_word_set = normalize_words(negative_words) positive_word_set = normalize_words(positive_words) chunks = split_review_chunks(content, REVIEW_CHUNK_SIZE) all_items = [] for chunk in chunks: llm_items = await review_chunk( content=chunk, types=types, require=require, ) llm_items = filter_positive_word_false_positives( llm_items, positive_word_set, ) llm_items = append_dictionary_results( llm_items, content=chunk, sensitive_words=sensitive_word_set, negative_words=negative_word_set, ) all_items.extend( item for item in llm_items if item.get("original", "").strip() and item.get("error", "").strip() and not _is_whitespace_only_change( item.get("original", ""), item.get("suggestion", "") ) ) return dedupe_and_filter(all_items, content) @app.post("/Review") async def review(request: ReviewRequest): return llm_text_stream_response(stream_review_content( content=request.content, types=request.types, require=request.require, sensitive_words=request.sensitive_words, negative_words=request.negative_words, positive_words=request.positive_words, )) # =============== 请求模型 =============== class AgentRequest(BaseModel): """Agent 请求模型""" query: Optional[Any] = None # 用户文本输入 file_text: Optional[str] = None # 文件文本输入 image: Optional[str] = None # 图片路径 audio: Optional[str] = None # 音频路径（支持 audio 或 asr） # 路由标志（可选，如果提供则跳过任务分类） route_flag: Optional[str] = None # 路由标志 history_message: Optional[Any] = None # 历史信息（支持字符串或列表格式） # 接收参数 top_k: Optional[Any] = None # rag 前top_k rag_prompt: Optional[Any] = None # rag 提示词模板（支持字符串、列表、字典等多种类型） image_kb_id: Optional[str] = None # 图库 知识库id image_file_id: Optional[str] = None # 图库 文件id text_kb_id: Optional[str] = None # 检索 知识库id text_file_id: Optional[str] = None # 检索 文件id # Socket.IO 格式所需字段 chat_id: Optional[str] = None # 聊天ID message_id: Optional[str] = None # 消息ID # 知识库配置 x_user_id: Optional[str] = "1" # 用户ID x_user_name: Optional[str] = "testuser" # 用户名称 x_role: Optional[str] = "admin" # 角色 class AgentResponse(BaseModel): """Agent 响应模型""" success: bool message: str data: Optional[Dict[str, Any]] = None error: Optional[str] = None class FeedbackClassifyRequest(BaseModel): """反馈分类请求模型""" feedback_text: str def _citation_display_filename(value: Any) -> Any: """仅转换文本溯源展示名，不修改 file_id 或实际文件路径。""" if not isinstance(value, str): return value return re.sub( r"(?i)\.(?:pptx|ppt)$", ".pdf", value.strip(), ) def normalize_source_citation_filenames( source_citation: Dict[str, Any], ) -> Dict[str, Any]: """将最终文本溯源中的 PPT/PPTX 展示后缀统一转换为 PDF。""" if not isinstance(source_citation, dict): return source_citation normalized: Dict[str, Any] = {} for resource_name, items in source_citation.items(): display_resource = _citation_display_filename(resource_name) if isinstance(items, list): normalized_items = [] for item in items: if not isinstance(item, dict): normalized_items.append(item) continue normalized_item = dict(item) for field in ("filename", "resource", "title"): if field in normalized_item: normalized_item[field] = _citation_display_filename( normalized_item[field] ) normalized_items.append(normalized_item) else: normalized_items = items # 极少数同名 .ppt/.pptx 会映射到同一 .pdf，保留全部切片。 if ( display_resource in normalized and isinstance(normalized[display_resource], list) and isinstance(normalized_items, list) ): normalized[display_resource].extend(normalized_items) else: normalized[display_resource] = normalized_items return normalized def format_stream_event(event_type: str, data: Dict[str, Any], chat_id: str = "", message_id: str = "", type: str = "chat:completion", ) -> str: """格式化流式事件输出为 Socket.IO 格式""" # 生成 message_id（如果没有提供） if not message_id: message_id = str(uuid.uuid4()) # 根据事件类型构建不同的数据结构 if event_type == "stream_content": # stream_content 事件使用 chat:completion 格式 content = data.get("content", "") event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "content": content } } } elif event_type == "function_end": # function_end 事件 event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "stepsBlueprint": data } } } elif event_type == "execution_complete": # execution_complete 事件 rag_result = data.get("rag_result", {}) graph_result = data.get("graph_result", []) final_result = data.get("final_result", {}) title = data.get("title", "") # 判断是否有有效的 RAG 结果（非空 dict） has_rag = isinstance(rag_result, dict) and bool(rag_result) # 判断是否有有效的图谱结果（非空 list，且至少一个元素的 nodes 非空） has_graph = ( isinstance(graph_result, list) and len(graph_result) > 0 and any(isinstance(item, dict) and item.get("nodes") for item in graph_result) ) # 只有当 RAG 或图谱有有效内容时，才构建 sourceCitation source_citation = None if has_rag or has_graph: source_citation = {} if has_rag: source_citation.update( normalize_source_citation_filenames(rag_result) ) if has_graph: source_citation["xxxx.graph"] = graph_result # 构建 event_data（title 由调用方异步计算后传入） if final_result: content = final_result.get("content", "") event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "content": content, }, "actions": final_result.get("actions", ""), "suggestedReplies": final_result.get("suggestedReplies", "") } } else: event_data = { "chat_id": chat_id, "message_id": message_id, "data": { "type": "chat:completion", "data": { "data": {}, "done": True, "title": title or "新对话", } } } # 仅在有引用来源时添加 sourceCitation 字段 if source_citation is not None: event_data["data"]["data"]["sourceCitation"] = source_citation else: # 其他事件 event_data = { "chat_id": chat_id, "message_id": message_id, "type": event_type, "data": { "type": event_type, "data": "" } } # Socket.IO 格式：["chat-events", {...}] socket_io_message = ["chat-events", event_data] return f"{json.dumps(socket_io_message, ensure_ascii=False)}\n" async def execute_workflow_chain( route_flag: str, route_params: Dict[str, Any], chat_id: str = "", message_id: str = "", checkpointer=None ) -> Dict[str, Any]: """ 执行工作流 Args: route_flag: 路由标志 route_params: 路由参数 chat_id: 聊天会话 ID message_id: 消息 ID checkpointer: LangGraph checkpointer 实例 Returns: 工作流执行结果 """ print("子工作流标志：", route_flag) print("子工作流参数：", route_params) if not route_flag: return {} extracted_text = route_params.get("extracted_text", "") combined_query = route_params.get("combined_query", "") history_message = route_params.get("history_message", "") workflow_info = WORKFLOW_CONFIG.get(route_flag) if not workflow_info: logger.warning(f"未知的路由标志: {route_flag}，使用默认的qa工作流") workflow_info = WORKFLOW_CONFIG.get("qa", WORKFLOW_CONFIG.get(list(WORKFLOW_CONFIG.keys())[0])) try: module = importlib.import_module(workflow_info["module"]) workflow_func = getattr(module, workflow_info["function"]) except (ImportError, AttributeError) as e: print(e) logger.error(f"无法导入工作流 {route_flag}: {e}") return {} print("zigongggggggggggggggggggggggg") try: import inspect sig = inspect.signature(workflow_func) params = list(sig.parameters.keys()) kwargs = { "extracted_text": extracted_text, "combined_query": combined_query, "history_message": history_message, "route_flag": route_flag, } if "chat_id" in params: kwargs["chat_id"] = chat_id if "message_id" in params: kwargs["message_id"] = message_id if "checkpointer" in params: kwargs["checkpointer"] = checkpointer result = await workflow_func(**kwargs) except Exception as e: print(e) logger.error(f"工作流执行失败: {e}") return {} return result or {} async def stream_main_agent_execution( raw_input: Dict[str, Any], route_flag: str = "", rag_prompt: str = "", chat_id: str = "", message_id: str = "" ) -> AsyncGenerator[str, None]: """ 流式执行主Agent并输出函数调用信息（使用装饰器事件） Args: raw_input: 原始输入数据 route_flag: 路由标志 rag_prompt: RAG 提示词 chat_id: 聊天会话 ID message_id: 消息 ID Yields: 格式化的流式事件字符串 """ if not chat_id: chat_id = str(uuid.uuid4()) if not message_id: message_id = str(uuid.uuid4()) stream_handler, event_queue = create_stream_event_handler() register_event_callback(stream_handler) try: execution_done = asyncio.Event() execution_error = None final_result = None checkpointer = None async def run_agent(): nonlocal execution_error, final_result, checkpointer route_flag_value: str = "" try: from checkpointer_config import checkpointer_manager checkpointer = await checkpointer_manager.get_async_checkpointer() main_agent = create_main_agent() initial_route_flag: str = route_flag if (route_flag and route_flag in VALID_ROUTE_FLAGS) else "" initial_state = { "raw_input": raw_input, "input_type": "", "has_query":False, "has_image": False, "has_audio": False, "has_file_text": False, "history_message":raw_input.get("history_message", ""), "extracted_text": "", "image_description": "", "asr_text": "", "file_text": raw_input.get("file_text", ""), "combined_query": "", "task_type": "", "task_classification_result": None, "workflow_result": None, "final_response": "", "error_message": "", "route_flag": initial_route_flag, "route_params": {} } main_result = await main_agent.ainvoke(initial_state) route_flag_value = (main_result.get("route_flag")or main_result.get("task_type")or "") route_params = main_result.get("route_params") or {"combined_query": main_result.get("combined_query", "") or main_result.get("extracted_text", ""),"extracted_text": main_result.get("extracted_text", "")} if "file_text" in raw_input: route_params["file_text"] = raw_input["file_text"] if "history_message" not in route_params and "history_message" in raw_input: route_params["history_message"] = filter_image_urls(raw_input["history_message"]) print("历史记录",route_params["history_message"]) print("用户问题", route_params["combined_query"]) print("ttttttt", route_params["extracted_text"]) sub_result = await execute_workflow_chain( route_flag=route_flag_value, route_params=route_params, chat_id=chat_id, message_id=message_id, checkpointer=checkpointer ) processed_content = "" if sub_result.get("response"): processed_content = sub_result.get("response", "") else: logger.warning("子agent没有返回response") processed_content = "" final_result = { "content": processed_content, "actions": sub_result.get("actions", []), "result_tag": sub_result.get("result_tag", ""), "suggestedReplies": sub_result.get("suggestedReplies", []) } except Exception as e: execution_error = e finally: execution_done.set() await event_queue.put({"type": "execution_complete"}) agent_task = asyncio.create_task(run_agent()) execution_complete_received = False has_streamed_content = False rag_result = [] graph_result = [] while True: try: event = await asyncio.wait_for(event_queue.get(), timeout=0.1) if event.get("type") == "execution_complete": execution_complete_received = True continue event_type = event.get("type") title = event.get("title", "") details = event.get("details", "") result = event.get("result") if event_type == "function_execution": logger.info(f"function_execution event - title: {title}, has_result: {result is not None}, result_type: {type(result)}") # 装饰器产生的无标题事件属于内部入参/返回值，不应在前端展示。 # 前端会把它们自动编号成“步骤 2”“步骤 5”，从而泄露检索原文 # 和未整理的回答草稿。 if not title.strip(): continue if title == "知识库搜索工具" and result and isinstance(result, dict): print("===============================", result) rag_result_raw = result.get("sourceCitation", {}) rag_result = {} for doc_name, chunks in rag_result_raw.items(): if isinstance(chunks, list): rag_result[doc_name] = [ {k: v for k, v in item.items() if k != "text"} if isinstance(item, dict) else item for item in chunks ] else: rag_result[doc_name] = chunks elif title == "图谱检索工具" and result and isinstance(result, dict): print("===============================", result) graph_result = result.get("xxxx.graph", []) else: yield format_stream_event("function_end", { "title": title, "details": details }, chat_id=chat_id, message_id=message_id) elif event_type == "function_error": yield format_stream_event("function_error", { "title": title, "error": details, "details": details }, chat_id=chat_id, message_id=message_id) elif event_type == "stream_content": content = event.get("content", "") if content: has_streamed_content = True yield format_stream_event("stream_content", { "content": content }, chat_id=chat_id, message_id=message_id) except asyncio.TimeoutError: if execution_complete_received: break if execution_done.is_set() and event_queue.empty(): break continue await agent_task if execution_error: yield format_stream_event("error", { "error": str(execution_error) }, chat_id=chat_id, message_id=message_id) else: title = await extract_conversation_title(final_result.get("content", "") if final_result else "") frontend_final_result = final_result if has_streamed_content and final_result: # 正文已按增量流式发送，完成事件只补充 actions 等元数据。 # 再携带完整 content 会被前端追加为第二份重复正文。 frontend_final_result = {**final_result, "content": ""} yield format_stream_event("execution_complete", { "final_result": frontend_final_result }, chat_id=chat_id, message_id=message_id) yield format_stream_event("execution_complete", { "rag_result": rag_result, "graph_result": graph_result, "title": title }, chat_id=chat_id, message_id=message_id) except Exception as e: yield format_stream_event("error", { "error": str(e) }, chat_id=chat_id, message_id=message_id) finally: unregister_event_callback(stream_handler) clear_current_stream_handler() # =============== API 接口 =============== @app.post("/api/agent/stream_process") async def stream_process_request(request: AgentRequest): """ 流式处理 Agent 请求，实时输出节点执行信息 不保存历史，调用端会自己保存历史对话信息 每次请求时创建新的主Agent实例 """ print(f"🔍 [调试] 接收到的原始请求: {request.model_dump_json(indent=2)}") print(f"[DEBUG] 收到请求 chat_id={request.chat_id!r}, message_id={request.message_id!r}, route_flag={request.route_flag!r}, query={str(request.query)[:100]}") # 1. 构建当前请求的输入数据 raw_input = {} if request.query: if isinstance(request.query, str): raw_input["query"] = request.query elif isinstance(request.query, list): for item in request.query: if item.get("type") == "text": raw_input["query"] = item.get("text") print(f"当前查询: {raw_input['query']}") if request.image: raw_input["image"] = request.image if request.audio: # audio 字段支持 audio 或 asr，统一使用 audio 键 raw_input["audio"] = request.audio # [新增] 提取 file_text if request.file_text: raw_input["file_text"] = request.file_text[:5000] # 历史消息预处理：解析、移除当前用户消息、序列化 if request.history_message: raw_input["history_message"] = preprocess_from_request(request.history_message, exclude_last=True) # 2. 验证至少有一个输入 if not raw_input: raise HTTPException( status_code=400, detail="至少需要提供一个输入：query、image 或 audio" ) # 3. 获取路由标志（如果提供，将跳过任务分类步骤） # 始终传递 route_flag，默认值为空字符串 route_flag = request.route_flag or "" print("打印agent对应的route_flag：") print(route_flag) rag_prompt = process_rag_prompt(request.rag_prompt) if request.rag_prompt else "" chat_id = request.chat_id or "" message_id = request.message_id or "" # 记录智能体使用情况 if route_flag and chat_id: from tools.agent_usage_statistics import record_agent_usage asyncio.create_task(record_agent_usage( chat_id=chat_id, route_flag=route_flag, message_id=message_id )) logger.info(f"记录智能体使用: chat_id={chat_id}, route_flag={route_flag}") # 4. 设置当前请求的用户配置和知识库参数（线程安全，支持多用户并发） # 如果请求中提供了这些参数，则使用请求中的值；否则使用默认值 set_request_user_config( x_user_id=request.x_user_id, x_user_name=request.x_user_name, x_role=request.x_role, text_kb_id=request.text_kb_id, text_file_id=request.text_file_id ) #5. 执行主Agent async def event_generator(): async for event in stream_main_agent_execution( raw_input=raw_input, route_flag=route_flag, rag_prompt=rag_prompt, chat_id=chat_id, message_id=message_id ): yield event return StreamingResponse(event_generator(), media_type="text/event-stream") @app.get("/api/download/{file_name}") async def download_file(file_name: str): """ 文件下载接口 Args: file_name: 文件名 Returns: 文件下载响应 """ pdf_path = os.path.join(PDF_DIR, file_name) word_path = os.path.join(WORD_DIR, file_name) file_path = pdf_path if os.path.exists(pdf_path) else word_path print(f"下载文件：{file_path}") # 检查文件是否存在 ext = os.path.splitext(file_name)[1].lower() media_type = { ".pdf": "application/pdf", ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ".doc": "application/msword", }.get(ext, "application/octet-stream") # 返回文件 return FileResponse( path=file_path, filename=file_name, media_type=media_type ) @app.post("/api/v1/feedback/classify") async def feedback_classify(request: FeedbackClassifyRequest): """ 反馈内容分类接口 功能： 1. 从反馈文本中提取设备名称 2. 根据设备名称反推系统名称 3. 对反馈内容进行分类 Args: request: 包含 feedback_text 字段的请求体 Returns: { "success": true, "system": "系统名称", "category": "分类结果" } 分类类别： - 信息过时：反映文档或手册中的信息已经过时，不再适用 - 步骤错误：反映操作步骤或流程存在错误 - 步骤缺失：反映缺少必要的操作步骤或流程说明 - 备件信息有误：反映备件型号、规格、数量等信息有误 - 安全提示不足：反映缺少必要的安全警示或注意事项 - 图示不清：反映图片、示意图不清晰或难以理解 - 其他：不属于以上类别 """ from tools.fault_statistics import analyze_feedback_and_classify result = await analyze_feedback_and_classify(request.feedback_text) if not result.get("success", False): raise HTTPException( status_code=500, detail=result.get("error", "分类失败") ) return result class PdfContentItem(BaseModel): page_idx: int bbox: Optional[str] = None text: Optional[str] = None text_level: Optional[int] = None img_path: Optional[str] = None table_body: Optional[str] = None class RequestWrapper(BaseModel): pdf_contents: List[PdfContentItem] class Base64ImageRequest(BaseModel): content: Optional[str] = Field( None, description="Optional text around the image, for example: 图3.1 船舶主发动机组成图（images/test.jpg）", ) filename: str = Field(..., description="Original image filename, for example test.png") image_base64: str = Field(..., description="Base64 image content, with or without data:image/... prefix") async def analyze_image_with_vlm(image_bytes: bytes, suffix: str) -> str: image_url = image_base64_to_data_url(image_bytes, suffix) async with VLM_SEMAPHORE: return await model_api.OpenaiAPI.open_api_vl_without_thinking(image_url) @app.post("/split_image") async def split_image(request_data: Base64ImageRequest): filename = safe_original_filename(request_data.filename) suffix = PathLib(filename).suffix.lower() if suffix not in {".jpg", ".jpeg", ".png", ".bmp", ".webp", ".gif", ".tif", ".tiff"}: raise HTTPException(status_code=400, detail="Unsupported image type") image_bytes, processed_suffix = prepare_split_image(request_data.image_base64, suffix) if not image_bytes: raise HTTPException(status_code=400, detail="decoded image is empty") temp_path = None try: with tempfile.NamedTemporaryFile(delete=False, suffix=processed_suffix) as tmp: tmp.write(image_bytes) temp_path = PathLib(tmp.name) ocr_data = await extract_image_text(temp_path) try: temp_path.unlink() except Exception as e: logger.warning(f"Failed to delete temp image {temp_path}: {e}") finally: temp_path = None content = (request_data.content or "").strip() ocr_full_text = (ocr_data.get("full_text") or "").strip() if not content and not ocr_full_text: vlm_text = await analyze_image_with_vlm(image_bytes, processed_suffix) image_content = "图片文本描述为：" + (vlm_text or "").strip() elif content and ocr_full_text: image_content = f"图片上下文内容为：{content}, {ocr_full_text}" elif content: image_content = "图片上下文内容为：" + content else: image_content = ocr_full_text image_content = "图片文件名为：" + filename + "，图片相关内容为：" + image_content return JSONResponse( { "code": 200, "message": "ok", "data": { "image_content": image_content, "filename": filename, }, } ) finally: if temp_path and temp_path.exists(): try: temp_path.unlink() except Exception as e: logger.warning(f"Failed to delete temp image {temp_path}: {e}") @app.post("/split_content_list") async def split_content_list(request_data: RequestWrapper): pdf_contents = request_data.pdf_contents processed_list = [] for item in pdf_contents: # 将 Pydantic 模型转换为字典，方便修改 item_dict = item.model_dump() # --- 核心逻辑：判断并添加 type 字段 --- item_type = None # 1. 如果 text 非空 -> type: "text" if item.text: item_type = "text" # 2. 如果 table_body 非空 -> type: "table" # 注意：如果 text 和 table_body 都有，根据代码顺序，table 会覆盖 text。 # 如果你的业务逻辑是互斥的或需要优先级，请调整此处顺序。 elif item.table_body: item_type = "table" # 3. 如果 img_path 非空 且 table_body 为空 -> type: "image" # 注意：上面用了 elif 判断 table_body，所以到这里 table_body 肯定为空， # 但为了逻辑清晰，显式写出条件。 elif item.img_path and not item.table_body: item_type = "image" # 如果匹配到了类型，写入字典 if item_type: item_dict["type"] = item_type processed_list.append(item_dict) slices = get_chunk_bbox(processed_list) slices_check = chunk_check(slices, 8000) return {"code": 200, "message": "ok", "data": {"slices": slices_check}} @app.post("/split_result") async def split_result( file: UploadFile = File(...), image_prefix: str = Form("/api/v1/knowledge/files/images/"), ): if not file.filename: raise HTTPException(status_code=400, detail="File name is missing") chunk_size = 1024 chunk_overlap = 100 filename = file.filename suffix = PathLib(filename).suffix.lower() # 获取文件后缀 temp_input_path: Optional[PathLib] = None converted_pdf_path: Optional[PathLib] = None logger.info("调用知识切分接口") try: # ? 创建临时文件时指定后缀名 with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_in: content = await file.read() logger.info(f"文件大小:{len(content)}") logger.info(f"前10字节:{content[:10]}") # if not content.startwith(b"%PDF"): # logger.error("不是合法PDF") # raise HTTPException(status_code=400,detail="Uploaded file is not a valid PDF") tmp_in.write(content) temp_input_path = PathLib(tmp_in.name) # 2. 根据文件类型处理 if suffix == ".pdf": logger.info("调用pdf切分方式") result_data = await process_pdf_file(temp_input_path, image_prefix,filename) if result_data is None: raise HTTPException(status_code=502, detail="PDF processing failed") elif suffix == ".docx": logger.info("调用docx切分方式") logger.info(f"转换 DOCX -> PDF: {filename}") # 执行转换 get_doc2pdf_converter().convert(str(temp_input_path), output_dir=str(DATA_DIR)) # ? 使用临时文件的基础名查找 PDF temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf" if not converted_pdf_path.exists(): raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}") result_data = await process_pdf_file(converted_pdf_path, image_prefix) if result_data is None: raise HTTPException(status_code=502, detail="DOCX to PDF processing failed") elif suffix in (".txt", ".md"): logger.info("调用txt切分方式") # 直接读取文本内容 async with aiofiles.open(temp_input_path, "r", encoding="utf-8", errors="ignore") as f: text_content = await f.read() contents = split_text_preserve_sentences(text_content) # 构造统一结构（无 images） slices = [] for ins in contents: slices.append({"content": ins, "positions": []}) result_data = {"slices": slices, "images": {}} elif suffix in [".xlsx"]: # converter.convert(str(temp_input_path), output_dir=str(DATA_DIR)) # temp_base_name = temp_input_path.stem # 如 tmp2bb0l4cn # converted_pdf_path = PathLib(DATA_DIR) / f"{temp_base_name}.pdf" # if not converted_pdf_path.exists(): # raise HTTPException(status_code=500, detail=f"PDF 转换失败: {converted_pdf_path}") logger.info("调用xlsx切分方式") result_data = await process_other_file(temp_input_path, image_prefix,filename) if result_data is None: raise HTTPException(status_code=502, detail="DOCX parse failed") elif suffix in [".ppt", ".pptx"]: logger.info("调用 PPT/PPTX 切分方式") # 1. 判断是否为旧版 .ppt 格式，如果是，则先进行转换 if suffix == ".ppt": logger.info(f"检测到旧版 PPT 文件，正在转换为 PPTX: {temp_input_path}") try: # 定义 LibreOffice 转换命令 command = [ 'libreoffice', '--headless', # 无头模式，不弹出图形界面 '--convert-to', 'pptx', '--outdir', os.path.dirname(temp_input_path), temp_input_path ] # 2. 使用 run_in_executor 将阻塞的同步转换操作放入线程池，避免阻塞异步事件循环 loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, lambda: subprocess.run(command, capture_output=True, text=True) ) # 3. 检查转换是否成功 if result.returncode != 0: logger.error(f"PPT 转换失败: {result.stderr}") raise HTTPException(status_code=500, detail="PPT to PPTX conversion failed") # 4. 转换成功后，更新输入路径为新生成的 .pptx 文件路径 temp_input_path = os.path.splitext(temp_input_path)[0] + ".pptx" logger.info(f"PPT 转换成功，新文件路径: {temp_input_path}") except Exception as e: logger.exception(f"PPT 转换过程发生异常: {e}") raise HTTPException(status_code=500, detail=f"PPT conversion error: {str(e)}") # 5. 无论是原本就是 .pptx，还是刚刚转换完成的 .pptx，都统一走后续处理逻辑 result_data = await process_other_file(temp_input_path, image_prefix, filename) if result_data is None: raise HTTPException(status_code=502, detail="PPTX parse failed") else: raise HTTPException( status_code=400, detail="Unsupported file type. Only PDF, DOCX, TXT, and MD are supported." ) return JSONResponse({"code": 200, "message": "ok", "data": result_data}) except requests.RequestException as e: logger.error(f"Split service request failed: {e}") raise HTTPException(status_code=502, detail=f"Split service error: {str(e)}") except Exception as e: logger.error(f"Internal error: {e}") raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") finally: # 清理临时文件 for path in [temp_input_path, converted_pdf_path]: if path and path.exists(): try: path.unlink() except Exception as e: logger.warning(f"Failed to delete temp file {path}: {e}") @app.get("/api/health") async def health_check(): """健康检查接口""" return { "status": "healthy", "message": "服务正常运行，每次请求时创建新的Agent实例" } @app.get("/") async def root(): """根路径""" return { "message": "Agent API 服务", "docs": "/docs" } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=9092)