浅聊一下，大模型应用架构 | 工程研发的算法修养系列（二）

def read_word_document(file_path): doc = docx.Document(file_path) paragraphs = [para.text.strip() for para in doc.paragraphs if para.text.strip()] return "\n".join(paragraphs) # 合并段落，形成完整文本 

欢迎大家来到IT世界,在知识的湖畔探索吧!

欢迎大家来到IT世界,在知识的湖畔探索吧!def split_text(text, chunk_size=100, chunk_overlap=10): text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n", "。", "？", "！"] ) return text_splitter.split_text(text) 

def load_embedding_model(model_name="moka-ai/m3e-base"): # 先下载模型（首次运行需联网） embeddings = HuggingFaceEmbeddings( model_name="moka-ai/m3e-base", # Hugging Face 仓库路径 model_kwargs={"device": "cpu"}, # 可选 GPU 加速 encode_kwargs={"normalize_embeddings": True} # 归一化向量 ) return embeddings

欢迎大家来到IT世界,在知识的湖畔探索吧!from langchain_community.vectorstores import FAISS
def store_to_vector_db(docs, db_path="faiss_index"): embeddings = load_embedding_model() vector_db = FAISS.from_documents(docs, embeddings) vector_db.save_local(db_path)

def search_similar_texts(query, vector_db, top_k=3): results = vector_db.similarity_search(query, k=top_k) return [r.page_content for r in results]

欢迎大家来到IT世界,在知识的湖畔探索吧!def fulltext_search(query): conn = sqlite3.connect(DB_FILE) cursor = conn.cursor() data = " OR ".join(jieba.cut(query)) cursor.execute("SELECT ori, bm25(documents) AS score FROM documents WHERE content MATCH ? ORDER BY score DESC LIMIT 3", (data,)) results = cursor.fetchall() conn.close() data = [item[0] for item in results] return data

def query_knowledge_base(user_query, index_path="faiss_index"): keyword_results = fulltext_search(user_query) vector_results = search_word_vector(user_query, load_vector_db(index_path)) return vector_results + keyword_results # 仅简单合并结果

欢迎大家来到IT世界,在知识的湖畔探索吧!def generate(prompt: str, model: str = "deepseek-r1:1.5b", stream: bool = False) -> str: url = "http://localhost:11434/api/generate" # ollama默认端口 payload = { "model": model, "prompt": prompt, "options": { "temperature": 0.7, # 控制生成随机性  "num_predict": 8192 # 最大生成长度 } } response = requests.post(url, json=payload) return response.json()["response"]

def run(self, task: str) -> str: # 工作记忆(Memory)：记录任务到交互历史 self.memory.add_message("user", task)
 # 规划模块(Planner)：分解任务创建执行计划 plan = self.planner.create_plan(task) # 提示模板简化：计划创建提示... self.memory.save_state("plan", plan)
 # 执行器(Executor)：准备执行计划 completed_steps = []
 for step in plan: step_id = step["step_id"] description = step["description"] tool_name = step.get("tool") print(f"执行步骤 {step_id}: {description}")
 if tool_name: # 推理引擎(LLM)：确定工具参数和使用方式 tools_description = self._format_tools_description() system_msg = self.system_prompt.format(tools_description=tools_description) # 提示模板(Prompts)：工具使用提示... messages = [ {"role": "system", "content": system_msg}, {"role": "user", "content": f"请帮我完成这个步骤：{description}。如果需要，请使用适当的工具。"} ]
 # 推理引擎(LLM)：生成响应 response = self.llm_engine.generate(messages) self.memory.add_message("assistant", response)
 # 观察处理器(Observer)：解析工具调用 tool_calls = self._parse_tool_calls(response)
 for tool_call in tool_calls: try: # 执行器(Executor)：实际调用工具 result = self.executor.execute_tool( tool_call["tool_name"],  tool_call["parameters"] )
 # 观察处理器(Observer)：处理和解释结果 observation = self.observer.process_result(description, result)
 # 工作记忆(Memory)：记录执行结果 step_result = { "step_id": step_id, "description": description, "tool_used": tool_call["tool_name"], "parameters": tool_call["parameters"], "result": result, "observation": observation } completed_steps.append(step_result) self.memory.add_message("system", f"工具执行结果: {result}")
 # 反馈循环(Feedback Loop)：评估行动效果并调整策略 remaining_steps = [s for s in plan if s["step_id"] not in [cs["step_id"] for cs in completed_steps]] feedback = self.feedback_loop.evaluate_and_adjust( task, completed_steps, observation, remaining_steps ) # 提示模板(Prompts)：反馈评估提示...
 # 规划模块(Planner)：根据反馈调整计划 if feedback.get("需要调整", False): new_plan = feedback.get("调整后的计划", []) plan = [s for s in completed_steps] + new_plan self.memory.save_state("plan", plan) print("计划已调整")
 except Exception as e: # 工作记忆(Memory)：记录错误信息 error_msg = f"执行步骤 {step_id} 时发生错误: {str(e)}" print(error_msg) self.memory.add_message("system", error_msg) else: # 非工具步骤处理 step_result = { "step_id": step_id, "description": description, "completed": True } completed_steps.append(step_result)
 # 推理引擎(LLM)：生成最终总结 summary_prompt = f""" 你刚刚帮助用户完成了以下任务：{task}  已完成的步骤： {json.dumps(completed_steps, ensure_ascii=False, indent=2)}  请提供一个简洁的总结，说明任务完成情况和主要结果。 """  messages = [{"role": "user", "content": summary_prompt}] summary = self.llm_engine.generate(messages)
 # 工作记忆(Memory)：记录总结 self.memory.add_message("assistant", summary) return summary

欢迎大家来到IT世界,在知识的湖畔探索吧!# Prompt templateprompt = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的旅游规划助手。你的任务是根据用户偏好创建详细的旅行计划。 使用可用的工具收集必要信息并创建全面的旅行策略。 始终考虑天气条件、交通选择、住宿可用性和当地景点。 提供详细的每日行程和费用估算。"""), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"),])
# Initialize the LLMdef get_llm() -> BaseChatModel: return ChatOpenAI( model="deepseek-chat", temperature=0.7, api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com/v1" # DeepSeek API endpoint )
# Create the agentdef create_agent(): try: # Fetch tools from server tools = create_tools_from_server_data() llm = get_llm() agent = create_openai_functions_agent(llm, tools, prompt) return AgentExecutor(agent=agent, tools=tools, verbose=True) except Exception as e: print(f"Error creating agent: {e}") return None
# Main function to create travel plandef create_travel_plan(preferences: TravelPreferences) -> TravelPlan: # 创建Agent agent_executor = await create_agent() # 初始化聊天历史 chat_history = [] # 创建初始提示 initial_prompt = f"""为{preferences.destination}从{preferences.start_date}到{preferences.end_date}创建一个旅行计划。 预算: ${preferences.budget} 兴趣爱好: {', '.join(preferences.interests)} 住宿偏好: {preferences.accommodation_preference}""" # 执行Agent response = await agent_executor.ainvoke({ "input": initial_prompt, }) # 从链输出中获取原始内容 content = response.get("output", "") # 创建一个简单的TravelPlan，将原始内容作为建议 return content 

# Initialize FastMCP servermcp = FastMCP("weather")
@mcp.tool()async def get_forecast(latitude: float, longitude: float) -> str: """获取某个地点的天气预报 Args: latitude: 纬度 longitude: 经度 """ # 省略具体获取天气的过程 # 此处可调用其他第三方气象平台接口 或 mock return mock_data
@mcp.tool()async def create_note(content: str) -> str: """创建一个新的备忘录并填入指定内容 Args: content: 要添加到新备忘录中的内容 """ # 先打开Notes应用 subprocess.run(['open', '-a', 'Notes']) # 使用AppleScript命令创建新笔记 applescript = f''' tell application "Notes" activate make new note at folder "Notes" with properties {{body:"{content}"}} end tell ''' subprocess.run(['osascript', '-e', applescript]) return "成功创建新笔记并填入内容"
if __name__ == "__main__": # Initialize and run the server mcp.run(transport='stdio')

欢迎大家来到IT世界,在知识的湖畔探索吧!{ "mcpServers": { "weather": { "command": "uv", "args": [ "--directory", "/Users/zq/weather", // 文件所在目录的绝对路径 "run", "weather.py" ] } }}

app = FastAPI()
async def get_weather_data(location: str, date: str) -> Dict[str, Any]: # 省略具体获取天气的过程 # 此处可调用其他第三方气象平台接口 或 mock return generate_mock_weather(location, date)
@app.post("/api/weather")async def get_weather(request: WeatherRequest): data = await get_weather_data(request.location, request.date) return {"status": "success", "data": data}
@app.get("/api/list_tools")async def list_tools(): """Return a list of all available tools in the MCP server""" tools = [ { "name": "get_weather", "description": "Get weather forecast for a specific location and date", "endpoint": "/api/weather", "method": "POST", "params": [ {"name": "location", "type": "string", "description": "The location to get weather for"}, {"name": "date", "type": "string", "description": "The date to get weather for"} ] } ] return {"status": "success", "tools": tools}
 if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) 

欢迎大家来到IT世界,在知识的湖畔探索吧!# Function to fetch tools from MCP serverasync def fetch_tools_from_server() -> List[Dict[str, Any]]: """Fetch available tools from the MCP server""" async with httpx.AsyncClient() as client: response = await client.get(f"{MCP_SERVER_URL}/api/list_tools") return response.json()["tools"]
def create_dynamic_tool_executor(tool_info: Dict[str, Any]) -> Callable: async def execute_api_call(*args, kwargs): payload = {} param_names = [p["name"] for p in tool_info["params"]] # Map args and kwargs to parameters for i, arg in enumerate(args): if i < len(param_names): payload[param_names[i]] = arg for key, value in kwargs.items(): if key in param_names: payload[key] = value # Execute request async with httpx.AsyncClient() as client: is_post = tool_info["method"] == "POST" request_kwargs = {"json" if is_post else "params": payload} method = client.post if is_post else client.get response = await method(f"{MCP_SERVER_URL}{tool_info['endpoint']}", request_kwargs)  if response.status_code == 200: return response.json()["data"] raise Exception(f"API call failed with status {response.status_code}: {response.text}") # Create synchronous wrapper with proper metadata def sync_executor(*args, kwargs): return asyncio.run(execute_api_call(*args, kwargs)) # Set function metadata sig_params = [inspect.Parameter(name=p["name"], kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,  annotation=param_type) for p in tool_info["params"]] sync_executor.__signature__ = inspect.Signature(parameters=sig_params) sync_executor.__name__ = tool_info["name"] sync_executor.__doc__ = tool_info["description"] return sync_executor# Function to create Tool objects from fetched tool datadef create_tools_from_server_data(tool_data: List[Dict[str, Any]]) -> List[Tool]: """Create Tool objects from fetched tool data""" tools = [] for tool_info in tool_data: # Create a dynamic tool executor for this tool executor = create_dynamic_tool_executor(tool_info) # Create a Tool object tools.append( Tool( name=tool_info["name"], func=executor, description=tool_info["description"] ) ) return tools