企业官网建设流程全解析

西安千度网站建设,做微信网站公司哪家好,网站能调用一些字体,买软件网站建设目标#xff1a;带你用 LangChain#xff08;Python#xff09;把第 2 课的五层架构落地成可运行的 Agent Demo。完成后你能理解 LangChain 的核心抽象#xff08;Agent / Tools / Memory / Chain / Prompt#xff09;#xff0c;并能搭出一个最小可用的“自动写周报” A…目标带你用 LangChainPython把第 2 课的五层架构落地成可运行的 Agent Demo。完成后你能理解 LangChain 的核心抽象Agent / Tools / Memory / Chain / Prompt并能搭出一个最小可用的“自动写周报” Agent。本课结构便于快速查阅课程前提与环境准备设计回顾五层如何映射到 LangChain最小可用 Agent 设计行为规范代码实现逐步解释测试与调试技巧安全、审计与生产化建议拓展练习一、前提与环境准备语言Python 3.11依赖建议虚拟环境langchain0.3.0langgraph0.2.0langchain-openai0.2.0langchain-experimental0.3.0DeepSeek 或其他 LLM SDK取决于你用哪个模型python-dotenv、tiktoken、faiss-cpu按需环境变量示例DEEPSEEK_API_KEY...LLM_PROVIDERdeepseekTESTING_MODEtrue安装示例pip install -r requirements.txt说明LangChain 定位为把 LLM 接入工具与记忆的框架它并非唯一方案但能把大部分工程实践抽象掉适合教学与快速落地。在最新版本中ReAct agent 已经迁移到 LangGraph 框架。二、把五层映射到 LangChainLLM 层LangChain 中的 LLM wrappersOpenAI、DeepSeek 等通过 ChatOpenAI 接口工具层Toolslangchain_core.tools.BaseTool自定义函数封装为工具记忆层Memorylangchain_community.chat_message_histories.ChatMessageHistoryConversationBufferMemory 已废弃控制层Controller / Orchestratorlanggraph.prebuilt.create_react_agent替代旧版 AgentExecutor接入层FastAPI / CLI / 企业 IM webhookLangChain 不强制但可与之集成要点真正控制 Agent 行为的部分在于你如何定义 Tools、Prompt 与 Agent Typetool-use agent、planner agent 等。在 LangChain 1.2.0 中ReAct agent 实现已迁移到 LangGraph。三、最小可用 Agent 设计功能需求最小可用拉取本周任务清单模拟从本地 JSON 或 mock API汇总完成项 / 进行中 / 风险项生成周报草稿指定模板将草稿写入本地文件dry-run 模式人工确认后写入把结果交给用户确认人工确认后可发送安全与工程约束所有写操作默认 dry-run除非用户确认所有操作记录审计日志谁、何时、调用了哪个工具、输出是什么使用 LangGraph 替代旧版 AgentExecutor 实现控制逻辑Agent 的核心能力在于把自然语言需求转为结构化动作比如 JSON{action:fetch_tasks,args: {...}}。四、代码实现重点分步解释下面给出简化但可运行的示例。注示例已更新为 LangChain 1.2.0 版本使用 LangGraph 实现 ReAct agent。依赖导入与 LLM 初始化支持 DeepSeek 举例 周报生成Agent实现 - 基于LangChain的五层架构 (更新版 - 使用LangGraph) import os from typing import Dict, Any, List from langchain_openai import ChatOpenAI from langchain_core.tools import BaseTool from langgraph.prebuilt import create_react_agent from memory.memory_manager import MemoryManager def create_weekly_report_agent(): 创建周报生成 Agent # LLM 初始化 - 使用环境变量检测是否为测试模式或特定LLM提供者 if os.getenv(TESTING_MODE, false).lower() true: # 在测试模式下使用模拟LLM # 使用一个模拟的ChatOpenAI实例来 ensure compatibility with bind_tools from langchain_core.messages import HumanMessage, AIMessage from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.language_models import BaseChatModel from typing import List, Optional import json class MockChatModel(BaseChatModel): Mock Chat Model for testing def _generate(self, messages, stopNone, run_managerNone, **kwargs): # Return a mock response based on the input content Mock response for testing return ChatResult(generations[ChatGeneration(messageAIMessage(contentcontent))]) property def _llm_type(self): return mock-chat def bind_tools(self, tools, **kwargs): # Return self to allow chaining return self llm MockChatModel() else: # 根据环境变量选择LLM提供者 llm_provider os.getenv(LLM_PROVIDER, openai).lower() if llm_provider deepseek: # 使用DeepSeek作为LLM提供者 (兼容OpenAI格式) llm ChatOpenAI( modelos.getenv(DEEPSEEK_MODEL, deepseek-chat), base_urlos.getenv(DEEPSEEK_API_BASE, https://api.deepseek.com), api_keyos.getenv(DEEPSEEK_API_KEY), temperaturefloat(os.getenv(LLM_TEMPERATURE, 0)) ) else: # 默认使用OpenAI llm ChatOpenAI( temperaturefloat(os.getenv(LLM_TEMPERATURE, 0)), api_keyos.getenv(OPENAI_API_KEY) ) # 导入工具 from tools.task_fetcher import tool_fetch_tasks from tools.metrics_summarizer import tool_summarize from tools.report_writer import tool_write # 创建工具列表 tools [tool_fetch_tasks, tool_summarize, tool_write] # 创建记忆管理器 memory_manager MemoryManager() # 使用记忆管理器的短期记忆 memory memory_manager.get_short_term_memory() # 创建 LangGraph ReAct agent agent create_react_agent(llm, tools) # 返回一个包含 agent 和 memory_manager 的字典便于后续访问 return { agent: agent, memory_manager: memory_manager }说明在 LangChain 1.2.0 中ReAct agent 创建方式已更新使用langgraph.prebuilt.create_react_agent替代旧版的initialize_agent。定义工具Tools工具要保持窄而稳。我们实现三个工具fetch_tasks、summarize_metrics、write_reportdry-run 写入本地真实写入需确认。# 模拟数据源 def fetch_tasks_from_mock(team: str, week: str) - Dict[str, Any]: 模拟从任务系统获取任务数据 # 实际应调用 Jira / Notion / API try: with open(sample6/mock_tasks.json, r, encodingutf-8) as f: data json.load(f) # 这里简单返回过滤结果 tasks [t for t in data if t[team] team and t[week] week] return {tasks: tasks} except FileNotFoundError: # 如果没有mock文件返回模拟数据 return { tasks: [ { id: TASK-001, title: 修复登录页面样式问题, status: completed, assignee: 张三, priority: high }, { id: TASK-002, title: 优化数据库查询性能, status: in_progress, assignee: 李四, priority: medium }, { id: TASK-003, title: 添加用户反馈功能, status: todo, assignee: 王五, priority: low } ] } # 把函数包装为 LangChain Tools class TaskFetchTool(BaseTool): name: str fetch_tasks description: str 按 team 和 week 获取任务列表输入格式: {\team\: \团队名\, \week\: \周\}返回 JSON 格式的任务列表 def _run(self, input_str: str) - str: try: # 尝试解析为JSON如果失败则尝试直接使用作为输入 try: input_data json.loads(input_str) except json.JSONDecodeError: # 如果输入不是JSON格式尝试解析为简单的键值对 # 这种情况可能发生在LangChain直接传递字符串而非JSON时 input_data {input: input_str} # 从输入数据中提取参数 team input_data.get(team) week input_data.get(week) # 如果直接获取不到尝试从input字段中解析 if not team or not week: input_text input_data.get(input, ) if isinstance(input_text, str): # 尝试从字符串中提取参数 import re # 匹配 team: value 或 team: value 等格式 if not team: team_match re.search(r?team?\s*[:]\s*([^]), input_text) if not team_match: team_match re.search(r?team?\s*[:]\s*(\w), input_text) if team_match: team team_match.group(1) if not week: week_match re.search(r?week?\s*[:]\s*([^]), input_text) if not week_match: week_match re.search(r?week?\s*[:]\s*(\w), input_text) if week_match: week week_match.group(1) if not team or not week: # 如果仍然无法提取返回错误信息 return json.dumps({ error: f无法从输入中提取team和week参数。期望格式: {{\team\: \团队名\, \week\: \周\}}实际输入: {input_str}, received_input: input_str }, ensure_asciiFalse) result fetch_tasks_from_mock(team, week) return json.dumps(result, ensure_asciiFalse) except Exception as e: # 记录错误到审计日志 audit_logger.log_action( user_idsystem, actiontool_execution, tool_nameself.name, inputs{input_str: input_str}, outputs{error: str(e)}, statuserror ) return json.dumps({error: str(e)}) async def _arun(self, input_str: str): raise NotImplementedError(fetch_tasks does not support async) # 创建工具实例 tool_fetch_tasks TaskFetchTool()说明在 LangChain 1.2.0 中工具定义需要继承BaseTool并实现_run方法同时需要添加类型注解以避免 Pydantic 验证错误。记忆层ChatMessageHistory 可选向量记忆RAG 记忆层实现包括短期记忆、长期记忆和记忆管理功能 import os from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings from langchain_community.embeddings import FakeEmbeddings, HuggingFaceEmbeddings from langchain_openai import OpenAI import json from typing import Dict, List, Any class MemoryManager: 记忆管理器整合短期记忆和长期记忆 def __init__(self): # 短期记忆聊天消息历史 self.short_term_memory ChatMessageHistory() # 长期记忆向量存储这里使用简单的实现实际应用中需要更复杂的管理 self.long_term_memory None # 根据环境变量决定使用真实或模拟嵌入 if os.getenv(TESTING_MODE, false).lower() true: # 在测试模式下使用模拟嵌入 from langchain_community.embeddings import FakeEmbeddings # 创建一个维度为10的模拟嵌入器 self.embeddings FakeEmbeddings(size10) elif os.getenv(EMBEDDING_PROVIDER, openai).lower() huggingface: # 使用HuggingFace嵌入提供者免费替代方案 self.embeddings HuggingFaceEmbeddings() else: # 默认使用OpenAI嵌入 # Check if OPENAI_API_KEY is available, otherwise use a default api_key os.getenv(OPENAI_API_KEY) if not api_key: # Use a fake embedding as fallback when no API key is provided from langchain_community.embeddings import FakeEmbeddings self.embeddings FakeEmbeddings(size10) else: self.embeddings OpenAIEmbeddings(openai_api_keyapi_key) # 审计日志 self.audit_log [] def add_to_short_term(self, input_text: str, output_text: str): 添加对话到短期记忆 self.short_term_memory.add_user_message(input_text) self.short_term_memory.add_ai_message(output_text) def get_short_term_memory(self): 获取短期记忆 return self.short_term_memory def add_to_long_term(self, text: str, metadata: Dict[str, Any] None): 添加信息到长期记忆 if self.long_term_memory is None: # 初始化向量存储 self.long_term_memory FAISS.from_texts([text], self.embeddings, metadatas[metadata or {}]) else: # 添加到现有向量存储 self.long_term_memory.add_texts([text], metadatas[metadata or {}]) def search_long_term(self, query: str, k: int 3) - List[str]: 从长期记忆中搜索相关信息 if self.long_term_memory is None: return [] docs self.long_term_memory.similarity_search(query, kk) return [doc.page_content for doc in docs] def log_audit(self, user_id: str, action: str, tool_name: str, inputs: Dict, outputs: Dict): 记录审计日志 log_entry { timestamp: str(__import__(datetime).datetime.now()), user_id: user_id, action: action, tool_name: tool_name, inputs: inputs, outputs: outputs } self.audit_log.append(log_entry) # 保存到文件 with open(audit_log.json, a, encodingutf-8) as f: f.write(json.dumps(log_entry, ensure_asciiFalse) \n) def get_audit_log(self) - List[Dict]: 获取审计日志 return self.audit_log说明在 LangChain 1.2.0 中ConversationBufferMemory已被废弃使用ChatMessageHistory作为短期记忆的替代方案。控制层使用 LangGraph 实现 控制层和接入层实现 包含工作流控制、安全检查和用户确认流程 from typing import Dict, Any import json from langchain_core.messages import HumanMessage class WorkflowController: 工作流控制器负责控制Agent执行流程 def __init__(self, agent): self.agent agent self.max_steps 8 # 最大步数限制 def execute_with_control(self, user_input: str) - Dict[str, Any]: 执行带控制的Agent流程 try: # 执行Agent - LangGraph expects input in the format {messages: [...]} # Create initial messages list with the user input inputs {messages: [HumanMessage(contentuser_input)]} # Invoke the LangGraph agent result self.agent.invoke(inputs) # Extract the output from the result if isinstance(result, dict) and messages in result: # Get the last message as the result messages result[messages] if messages: last_message messages[-1] # Get the last message result_text last_message.content else: result_text str(result) elif isinstance(result, dict): result_text result.get(output, str(result)) else: result_text str(result) # Record to memory manager (if exists) if hasattr(self.agent, memory_manager): self.agent.memory_manager.add_to_short_term(user_input, result_text) return { status: success, result: result_text, steps_used: len(self._get_steps_from_agent()) if hasattr(self.agent, _get_steps_from_agent) else unknown } except Exception as e: return { status: error, error_message: str(e), error_type: type(e).__name__ } def _get_steps_from_agent(self): 从Agent获取执行步骤如果支持 # 这里是一个简化实现实际可能需要根据具体的Agent类型获取步骤信息 return [] class UserConfirmationHandler: 用户确认处理器处理需要人工确认的操作 staticmethod def prepare_confirmation_payload(agent_result: str) - Dict[str, Any]: 准备确认载荷 # 解析Agent结果提取需要确认的信息 return { draft_content: agent_result, actions: [write_report], # 假设Agent建议执行写报告操作 } staticmethod def confirm_and_execute(confirm_payload: Dict[str, Any], tools): 处理用户确认并执行操作 if confirm_payload.get(confirmed, False): # 执行实际写入操作 if write_report in confirm_payload.get(actions, []): write_tool tools.get(write_report) if write_tool: # 从载荷中获取写入参数 write_params { report_md: confirm_payload.get(draft_content, ), path: confirm_payload.get(output_path, weekly_report.md), dry_run: False # 实际写入非dry run } result write_tool._run(json.dumps(write_params)) return json.loads(result) return {status: cancelled, message: 用户未确认执行}说明在 LangChain 1.2.0 中控制逻辑通过 LangGraph 实现输入格式为{messages: [HumanMessage(contentuser_input)]}。使用 Agent一个典型交互流程 自动写周报 Agent - 基于 LangChain 的五层架构实现 import os from agents.weekly_report_agent import create_weekly_report_agent from utils.controller import WorkflowController, UserConfirmationHandler import json from dotenv import load_dotenv # 加载环境变量 load_dotenv() def main(): print(自动写周报 Agent 启动中...) # 显示当前使用的LLM提供者 llm_provider os.getenv(LLM_PROVIDER, openai).lower() print(f使用 LLM 提供者: {llm_provider}) # 创建 Agent agent_components create_weekly_report_agent() agent agent_components[agent] # 创建工作流控制器 controller WorkflowController(agent) # 示例输入 user_input 请帮我生成公司后端组 2025-W50 的周报草稿 包含完成项、进行中项、风险与下周计划 调用任务系统获取任务teambackend, week2025-W50 拉取关键指标DAU、错误率 先做 dry-run生成草稿并给出建议等待我确认是否写入 print(用户输入:, user_input) print(\n正在生成周报...) # 执行带控制的流程 result controller.execute_with_control(user_input) if result[status] success: print(周报生成成功) print(结果:, result[result]) # 准备用户确认 confirmation_payload UserConfirmationHandler.prepare_confirmation_payload(result[result]) print(\n需要您的确认:) print(f草稿内容预览: {confirmation_payload[draft_content][:200]}...) # 模拟用户确认 print(\n用户已确认执行实际写入...) final_result UserConfirmationHandler.confirm_and_execute( {**confirmation_payload, confirmed: True}, {write_report: __import__(tools.report_writer).tool_write} ) print(最终结果:, final_result) else: print(周报生成失败:, result[error_message]) if __name__ __main__: main()说明在实际运行中Agent 会调用工具fetch_tasks、summarize_metrics、write_report并把输出记录到 memory。LangGraph agent 的输入格式与旧版有所不同需要使用正确的格式。人工确认与实际写入Agent 生成草稿后我们在外部UI/CLI提示用户确认。如果确认则调用 write_report 的 dry_runFalseconfirm_payload { report_md: agent 生成的 markdown 字符串, path: weekly_backend_2025-W50.md, dry_run: False } # 直接调用工具或通过 agent 执行特定 action write_result tool_write._run(json.dumps(confirm_payload)) print(write_result)这种做法可以强制区分建议与执行是工程上常见的安全模式。五、测试与调试技巧本地 mock 是关键先用 mock_tasks.json、mock BI 函数调通流程再接真正 API。打开 LLM 输出日志温度设为 0 或 0.2 提高稳定性在开发调高温度可看更多候选结果。增量构建 Tools先只实现 fetch_tasks generate_report不写文件保证逻辑再加入写工具、确认流程。加入观测Observability每次工具调用记录到日志时间、输入、输出、调用者。防死循环max_iterations 检测没有新操作或重复操作超过阈值。测试模式使用 TESTING_MODEtrue 环境变量启用模拟 LLM避免消耗真实 API 调用。六、安全、审计与生产化建议必读权限边界所有写权限需基于用户身份与审批策略。写操作默认 dry-run。审计日志记录每一步的 LLM 响应、工具输入输出、执行者系统/用户、时间戳。白名单工具在生产中只暴露有限工具集合禁止 arbitrary code execution除非有严格审计。速率与并发限制防止自动化任务造成系统冲击或滥用。回滚策略写操作应能回滚或保留版本如写入 Notion 之前先创建草稿副本。人机协同界面在关键点提示用户例如发送给主管前再确认并展示差异/摘要。API 密钥管理使用环境变量存储敏感信息避免硬编码。七、常见进阶与拓展建议练习使用 VectorDBChroma/FAISS将历史周报与模板作为长期记忆做 RAG 提升准确性。把 Agent 拆成两个角色PlanAgent做任务分解和 ExecAgent调用工具执行实现多 Agent 协作。为 fetch_tasks 添加增量查询只查询自上次周报以来的新任务并保留任务状态到短期记忆。在写入前做自动合规检查 Agent检查是否包含敏感信息或承诺。将 Agent 托管为后端服务FastAPI并对接企业 IM飞书/Slack作为接入层支持异步执行与通知。练习题强烈建议动手用 mock 数据把上面 Demo 跑通生成一个 markdown 周报文件并人工确认写入。改造工具使 fetch_tasks 真实调用你常用的 issue trackerJira/Notion/GitHub。为 Agent 增加任务失败重试 降级策略比如 metrics 查询失败时用上次缓存的数据。八、小结LangChain 把 LLM、Tools、Memory、LangGraph Agent 抽象成工程化构件便于把说话的模型变成会做事的智能体。工程核心不在于模型本身而在于如何设计窄而稳的工具、如何保存并检索记忆、如何在控制层避免失控、如何通过接入层安全地进入业务流程。在 LangChain 1.2.0 中ReAct agent 已迁移到 LangGraph 框架API 有所变化但核心理念不变。实践建议先做 mock dry-run再逐步放开真实读写权限保持日志与人工确认点。