所有文章

引言 AI Agent的强大能力来自于其工作流编排能力——将复杂任务分解为多个步骤，并智能地协调执行。从简单的链式调用到复杂的多Agent协作，工作流编排是Agent系统的核心。本文将深入探讨主流的Agent编排框架和实战技巧。 LangChain Chains 基础Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import OpenAI # 创建LLM llm = OpenAI(temperature=0) # 创建Prompt模板 prompt_template = PromptTemplate( input_variables=["product"], template="为{product}写一段吸引人的产品描述。" ) # 创建Chain chain = LLMChain(llm=llm, prompt=prompt_template) # 运行 description = chain.run(product="智能手表") print(description) Sequential Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from langchain.chains import SequentialChain # Chain 1: 生成产品名称 name_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_type"], template="为{product_type}产品想一个创意名称，只返回名称。" ), output_key="product_name" ) # Chain 2: 生成Slogan slogan_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name"], template="为{product_name}写一句简短有力的广告语。" ), output_key="slogan" ) # Chain 3: 生成完整描述 description_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name", "slogan"], template="产品名称：{product_name}\n广告语：{slogan}\n请基于以上信息写一段100字的产品描述。" ), output_key="description" ) # 组合Chain overall_chain = SequentialChain( chains=[name_chain, slogan_chain, description_chain], input_variables=["product_type"], output_variables=["product_name", "slogan", "description"] ) # 执行 result = overall_chain("智能手表") print(result) # { # 'product_name': 'TimePulse', # 'slogan': 'TimePulse - 让时间更有价值', # 'description': '...' # } Conditional Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 from langchain.chains import TransformChain # 条件判断函数 def categorize_price(inputs: dict) -> dict: price = inputs.get('price', 0) if price < 100: category = "low" elif price < 500: category = "medium" else: category = "high" return {"price_category": category} # 条件Chain price_categorize_chain = TransformChain( transform=categorize_price, input_variables=["price"], output_variables=["price_category"] ) # 不同价格段的不同处理 low_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是经济实惠的选择，适合预算有限的用户。写一段强调性价比的描述。" ) ) high_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是高端产品，强调其品质和独特价值。写一段描述。" ) ) from langchain.chains import RouterChain # 路由Chain router_chain = RouterChain( chains={ "low": low_price_chain, "medium": medium_price_chain, "high": high_price_chain }, default_chain=medium_price_chain ) LCEL (LangChain Expression Language) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 使用LCEL构建Chain prompt = PromptTemplate.from_template( "Tell me a joke about {topic}" ) # 使用管道操作符 (|) chain = ( prompt | llm | StrOutputParser() ) # 等价于 chain = prompt | llm | StrOutputParser() # 执行 result = chain.invoke({"topic": "programming"}) print(result) # 复杂的LCEL示例 from langchain_community.utilities import WikipediaSearch wiki_search = WikipediaSearch() research_chain = ( { "context": lambda x: wiki_search.run(x["topic"]), "topic": RunnablePassthrough() } | PromptTemplate.from_template( "Topic: {topic}\n\nResearch: {context}\n\nBased on the research, explain {topic} in simple terms." ) | llm | StrOutputParser() ) LangChain Agents ReAct Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import OpenAI # 定义工具 def search_tool(query: str) -> str: """搜索工具""" # 实际应用中调用搜索API return f"搜索'{query}'的结果：..." def calculator_tool(expression: str) -> str: """计算器工具""" try: result = eval(expression) return f"计算结果：{result}" except: return "计算错误" tools = [ Tool( name="Search", func=search_tool, description="用于搜索网络信息，输入应该是搜索查询" ), Tool( name="Calculator", func=calculator_tool, description="用于数学计算，输入应该是数学表达式" ) ] # 创建Agent llm = OpenAI(temperature=0) prompt = PromptTemplate.from_template( """Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: {agent_scratchpad}""" ) agent = create_react_agent( llm=llm, tools=tools, prompt=prompt ) # 创建Agent执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5 ) # 执行 result = agent_executor.invoke({ "input": "苹果公司现在的股价是多少？如果我有100股，总价值多少？" }) print(result["output"]) Custom Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain_openai import BaseOpenAI class CustomAgent(BaseSingleActionAgent): """自定义Agent""" llm: BaseOpenAI tools: list[Tool] @property def input_keys(self): return ["input"] def plan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """规划下一步行动""" user_input = kwargs["input"] # 构建思考过程 thoughts = "" for action, observation in intermediate_steps: thoughts += f"Action: {action.tool}\n" thoughts += f"Input: {action.tool_input}\n" thoughts += f"Observation: {observation}\n" # 让LLM决定下一步 prompt = f""" 输入: {user_input} 之前的步骤: {thoughts} 可用工具: {[tool.name for tool in self.tools]} 请决定下一步行动，格式为： Action: [工具名称] Input: [工具输入] 或如果已完成： Final Answer: [最终答案] """ response = self.llm.predict(prompt) # 解析响应 if "Final Answer:" in response: final_answer = response.split("Final Answer:")[-1].strip() return AgentAction( tool="FINAL", tool_input=final_answer, log=response ), final_answer else: # 提取Action和Input action_line = [l for l in response.split("\n") if "Action:" in l][0] input_line = [l for l in response.split("\n") if "Input:" in l][0] tool_name = action_line.split("Action:")[-1].strip() tool_input = input_line.split("Input:")[-1].strip() return AgentAction( tool=tool_name, tool_input=tool_input, log=response ), "" async def aplan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """异步规划""" return self.plan(intermediate_steps, **kwargs) # 使用自定义Agent custom_agent = CustomAgent( llm=OpenAI(temperature=0), tools=tools ) agent_executor = AgentExecutor( agent=custom_agent, tools=tools ) result = agent_executor.invoke({"input": "查询北京今天的天气"}) AutoGPT模式 基础AutoGPT实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 from typing import List, Dict import json class AutoGPTAgent: """AutoGPT风格的Agent""" def __init__( self, name: str, role: str, goals: List[str], llm, tools: Dict[str, callable] ): self.name = name self.role = role self.goals = goals self.llm = llm self.tools = tools self.memory = [] self.task_list = [] def think(self) -> Dict: """思考下一步行动""" prompt = f""" Name: {self.name} Role: {self.role} Goals: {', '.join(self.goals)} Memory: {self.format_memory()} Current Tasks: {self.format_tasks()} 请决定下一步行动。返回JSON格式： {{ "thought": "思考过程", "reasoning": "推理过程", "plan": "计划", "criticism": "自我批评", "action": "行动名称", "action_input": "行动输入" }} """ response = self.llm.generate(prompt) try: return json.loads(response) except: return { "thought": "解析错误", "action": "finish", "action_input": "" } def execute(self, action: str, action_input: str) -> str: """执行行动""" if action == "finish": return "任务完成" if action in self.tools: result = self.tools[action](action_input) # 记录到记忆 self.memory.append({ "action": action, "input": action_input, "result": result }) return result else: return f"未知行动: {action}" def format_memory(self) -> str: """格式化记忆""" if not self.memory: return "No memories yet." return "\n".join([ f"- {m['action']}: {m['input']} -> {m['result'][:100]}" for m in self.memory[-5:] ]) def format_tasks(self) -> str: """格式化任务列表""" if not self.task_list: return "No tasks." return "\n".join([ f"{i+1}. {task}" for i, task in enumerate(self.task_list) ]) def run(self, max_iterations: int = 10) -> str: """运行Agent""" for i in range(max_iterations): # 思考 thought_process = self.think() print(f"\n=== Iteration {i+1} ===") print(f"Thought: {thought_process['thought']}") print(f"Reasoning: {thought_process['reasoning']}") print(f"Plan: {thought_process['plan']}") print(f"Criticism: {thought_process['criticism']}") # 执行 action = thought_process['action'] action_input = thought_process['action_input'] result = self.execute(action, action_input) print(f"Action: {action}") print(f"Result: {result[:200]}") # 检查是否完成 if action == "finish": return result return "达到最大迭代次数" # 使用示例 def search_web(query: str) -> str: """搜索网络""" return f"搜索'{query}'的结果..." def write_file(content: str) -> str: """写入文件""" return "文件已写入" def read_file(filename: str) -> str: """读取文件""" return f"文件{filename}的内容..." tools = { "search": search_web, "write": write_file, "read": read_file } agent = AutoGPTAgent( name="Researcher", role="AI研究员", goals=["研究最新AI技术", "生成研究报告"], llm=OpenAI(temperature=0), tools=tools ) result = agent.run() print(result) BabyAGI模式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 class BabyAGI: """BabyAGI实现""" def __init__( self, objective: str, llm, tools: Dict[str, callable], max_iterations: int = 10 ): self.objective = objective self.llm = llm self.tools = tools self.max_iterations = max_iterations self.task_list = [] self.completed_tasks = [] def create_initial_tasks(self) -> List[str]: """创建初始任务列表""" prompt = f""" 目标: {self.objective} 请为这个目标创建一个任务列表。返回JSON数组格式： ["任务1", "任务2", "任务3"] """ response = self.llm.generate(prompt) try: tasks = json.loads(response) return tasks except: return ["研究相关资料", "分析问题", "制定方案"] def prioritize_tasks(self) -> List[str]: """任务优先级排序""" if not self.task_list: return [] prompt = f""" 目标: {self.objective} 当前任务列表: {json.dumps(self.task_list, ensure_ascii=False)} 已完成任务: {json.dumps(self.completed_tasks[-5:], ensure_ascii=False)} 请根据当前情况重新排列任务优先级。 返回JSON数组格式（从高到低）： ["任务1", "任务2", ...] """ response = self.llm.generate(prompt) try: return json.loads(response) except: return self.task_list def execute_task(self, task: str) -> str: """执行任务""" prompt = f""" 目标: {self.objective} 任务: {task} 请执行这个任务并返回结果。 如果需要使用工具，请说明： - search: 搜索信息 - calculate: 计算数据 - write: 写入内容 """ response = self.llm.generate(prompt) # 记录完成的任务 self.completed_tasks.append({ "task": task, "result": response }) return response def run(self) -> Dict: """运行BabyAGI""" # 创建初始任务 self.task_list = self.create_initial_tasks() for i in range(self.max_iterations): if not self.task_list: print("所有任务已完成！") break # 优先级排序 self.task_list = self.prioritize_tasks() # 执行第一个任务 current_task = self.task_list[0] print(f"\n=== 迭代 {i+1} ===") print(f"当前任务: {current_task}") result = self.execute_task(current_task) print(f"执行结果: {result[:200]}") # 从列表中移除 self.task_list.pop(0) return { "objective": self.objective, "completed_tasks": self.completed_tasks } # 使用 baby_agi = BabyAGI( objective="研究并总结2024年AI大模型的最新进展", llm=OpenAI(temperature=0), tools=tools ) result = baby_agi.run() CrewAI多Agent协作 Crew定义 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 from crewai import Agent, Task, Crew, Process # 定义Agent researcher = Agent( role='研究员', goal='研究最新的AI技术趋势', backstory="""你是一位经验丰富的AI研究员， 专注于追踪和分析最新的AI技术发展""", verbose=True, tools=[search_tool, wikipedia_tool] ) writer = Agent( role='技术作家', goal='将复杂的技术内容转化为易懂的文章', backstory="""你是一位技术写作专家， 擅长将技术细节转化为吸引人的内容""", verbose=True ) reviewer = Agent( role='内容审核员', goal='确保内容准确、完整、有价值', backstory="""你是一位资深的内容审核专家， 对技术内容的质量有极高要求""", verbose=True ) # 定义任务 research_task = Task( description="""研究2024年大语言模型的最新发展， 包括GPT-4、Claude、Gemini等模型的更新""", expected_output='详细的研究报告，包含关键发现和技术突破', agent=researcher ) write_task = Task( description="""基于研究报告，撰写一篇关于2024年LLM发展的技术文章。 文章应该面向技术读者，但保持通俗易懂""", expected_output='结构完整、内容丰富的技术文章（1000-1500字）', agent=writer ) review_task = Task( description="""审核技术文章，确保： 1. 技术准确性 2. 内容完整性 3. 可读性 4. 价值性 提供修改建议和最终评价""", expected_output='详细的审核报告，包含修改建议和最终评分', agent=reviewer ) # 创建Crew tech_crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, write_task, review_task], process=Process.sequential, # 顺序执行 verbose=True ) # 执行 result = tech_crew.kickoff() print(result) 并行Process 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 # 定义并行任务 crew_parallel = Crew( agents=[agent1, agent2, agent3], tasks=[task1, task2, task3], process=Process.parallel, # 并行执行 verbose=True ) # 或者使用层级Process crew_hierarchical = Crew( agents=[manager_agent, worker_agent1, worker_agent2], tasks=[manager_task, worker_task1, worker_task2], process=Process.hierarchical, # 层级执行 manager_llm=OpenAI(temperature=0), verbose=True ) Agent编排框架对比 LangGraph 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 from langgraph.graph import StateGraph, END from typing import TypedDict # 定义状态 class AgentState(TypedDict): input: str research: str draft: str review: str final: str # 创建图 workflow = StateGraph(AgentState) # 添加节点 def research_node(state: AgentState) -> AgentState: result = researcher_agent.run(state["input"]) return {**state, "research": result} def write_node(state: AgentState) -> AgentState: result = writer_agent.run(state["research"]) return {**state, "draft": result} def review_node(state: AgentState) -> AgentState: result = reviewer_agent.run(state["draft"]) return {**state, "review": result} # 添加节点到图 workflow.add_node("researcher", research_node) workflow.add_node("writer", write_node) workflow.add_node("reviewer", review_node) # 添加边 workflow.set_entry_point("researcher") workflow.add_edge("researcher", "writer") workflow.add_edge("writer", "reviewer") workflow.add_edge("reviewer", END) # 编译图 app = workflow.compile() # 执行 result = app.invoke({"input": "研究AI最新进展"}) Semantic Kernel 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion from semantic_kernel.planning import SequentialPlanner # 初始化Kernel kernel = Kernel() kernel.add_chat_service( "chat-gpt", OpenAIChatCompletion("gpt-4", api_key="...") ) # 定义技能（Skill） from semantic_kernel.skill_definition import sk_function class ResearchSkills: @sk_function(description="搜索信息") def search(self, query: str) -> str: return f"搜索'{query}'的结果..." @sk_function(description="总结内容") def summarize(self, content: str) -> str: return f"总结: {content[:100]}..." # 注册技能 kernel.import_skill(ResearchSkills(), skill_name="research") # 创建计划器 planner = SequentialPlanner(kernel) # 执行计划 ask = "研究2024年AI技术进展并生成报告" plan = await planner.create_plan_async(ask) result = await plan.invoke_async(kernel) print(result) 实战案例 案例一：智能研究报告生成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 class ReportGenerator: """智能报告生成器""" def __init__(self): self.llm = OpenAI(temperature=0) self.tools = self._init_tools() def _init_tools(self) -> Dict[str, callable]: return { "search": self._search, "analyze": self._analyze, "write": self._write, "format": self._format } def generate_report(self, topic: str) -> str: """生成报告""" # 阶段1：研究 research_data = self._research_stage(topic) # 阶段2：分析 analysis = self._analysis_stage(research_data) # 阶段3：撰写 draft = self._writing_stage(analysis) # 阶段4：审阅 final_report = self._review_stage(draft) return final_report def _research_stage(self, topic: str) -> Dict: """研究阶段""" # 生成研究计划 plan = self.llm.generate(f""" 为"{topic}"创建一个研究计划， 包含需要研究的关键点。 """) # 执行研究 research_data = {} for key_point in plan.split('\n'): if key_point.strip(): result = self._search(key_point.strip()) research_data[key_point.strip()] = result return research_data def _analysis_stage(self, data: Dict) -> str: """分析阶段""" prompt = f""" 分析以下研究数据： {json.dumps(data, ensure_ascii=False, indent=2)} 提供关键发现和洞察。 """ return self.llm.generate(prompt) def _writing_stage(self, analysis: str) -> str: """撰写阶段""" prompt = f""" 基于以下分析，撰写一份专业的研究报告： {analysis} 报告应该包含： 1. 执行摘要 2. 背景介绍 3. 主要发现 4. 结论和建议 """ return self.llm.generate(prompt) def _review_stage(self, draft: str) -> str: """审阅阶段""" prompt = f""" 审阅以下报告草稿： {draft} 提供改进建议并进行必要的修改。 """ reviewed = self.llm.generate(prompt) return reviewed def _search(self, query: str) -> str: """搜索实现""" # 实际调用搜索API return f"关于'{query}'的搜索结果..." 案例二：客户服务自动化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 class CustomerServiceAgent: """客户服务Agent""" def __init__(self): self.llm = OpenAI(temperature=0.7) self.knowledge_base = self._load_kb() self.conversation_history = {} def handle_customer_query( self, customer_id: str, query: str ) -> str: """处理客户查询""" # 获取历史对话 history = self.conversation_history.get(customer_id, []) # 检索相关知识 relevant_docs = self._retrieve_knowledge(query) # 理解意图 intent = self._classify_intent(query) # 生成响应 response = self._generate_response( query=query, intent=intent, history=history, knowledge=relevant_docs ) # 更新历史 history.append({"role": "user", "content": query}) history.append({"role": "assistant", "content": response}) self.conversation_history[customer_id] = history[-10:] return response def _classify_intent(self, query: str) -> str: """分类意图""" prompt = f""" 分类以下客户查询的意图： 查询: {query} 可能的意图： 1. 产品咨询 2. 订单查询 3. 投诉建议 4. 售后服务 5. 其他 只返回意图名称。 """ return self.llm.generate(prompt).strip() def _retrieve_knowledge(self, query: str) -> List[str]: """检索相关知识""" # 使用向量搜索 # 简化示例 return [ doc for doc in self.knowledge_base if any(word in doc.lower() for word in query.lower().split()) ][:3] def _generate_response( self, query: str, intent: str, history: List[Dict], knowledge: List[str] ) -> str: """生成响应""" history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in history[-5:] ]) knowledge_text = "\n".join(knowledge) prompt = f""" 意图: {intent} 知识库: {knowledge_text} 对话历史: {history_text} 客户查询: {query} 请提供专业、友好的回复。 """ return self.llm.generate(prompt) 总结 AI Agent工作流编排是构建复杂AI应用的关键技术。从简单的LangChain Chains到复杂的Multi-Agent Systems，不同的框架和模式适用于不同的场景。 ...

引言 去中心化身份（Decentralized Identity，简称DID）是WEB3的核心基础设施之一。它让用户完全掌控自己的身份数据，不再依赖中心化的身份提供商。本文将深入探讨DID的技术原理、W3C标准、可验证凭证（VC）以及如何构建生产级的去中心化身份系统。 DID基础概念 传统身份系统的问题 中心化身份平台 ├── Google账号 │ └── 谷歌掌握所有数据 ├── 微信账号 │ └── 腾讯掌握所有数据 └── 支付宝账号 └── 蚂蚁掌握所有数据 问题： - 数据孤岛：各平台数据不互通 - 隐私泄露：中心化服务器易被攻击 - 审查风险：平台可随时封禁账号 - 数据滥用：平台可擅自使用用户数据 去中心化身份的优势 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 // DID架构 interface DIDArchitecture { user: { control: "complete", // 用户完全控制 portable: true, // 身份可跨平台使用 privacy: "enhanced" // 隐私保护 }, verifier: { trust: "decentralized", // 去中心化信任 cost: "low" // 验证成本低 }, issuer: { efficiency: "high", // 发行效率高 revocation: "easy" // 撤销机制简单 } } W3C DID标准 DID结构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 // DID URL格式 did:method:specific-idstring // 示例 did:ethr:0x5a2e... // Ethereum DID did:sol:1234... // Solana DID did:web:example.com // Web DID did:key:z6Mk... // Key DID // 解析DID interface DID { method: string // 方法名（ethr, sol, web, key等） id: string // 特定方法的标识符 } DID文档 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 { "@context": [ "https://www.w3.org/ns/did/v1" ], "id": "did:ethr:0x5a2e...", "verificationMethod": [ { "id": "did:ethr:0x5a2e...#controller", "type": "EcdsaSecp256k1RecoveryMethod2020", "controller": "did:ethr:0x5a2e...", "blockchainAccountId": "0x5a2e...@eip155:1" } ], "authentication": [ "did:ethr:0x5a2e...#controller" ], "assertionMethod": [ "did:ethr:0x5a2e...#controller" ], "capabilityDelegation": [ "did:ethr:0x5a2e...#controller" ], "capabilityInvocation": [ "did:ethr:0x5a2e...#controller" ], "keyAgreement": [ { "id": "did:ethr:0x5a2e...#keyAgreement", "type": "X25519KeyAgreementKey2019", "controller": "did:ethr:0x5a2e...", "publicKeyBase58": "H3C2AVvLMv6gmMNam3uVAjZpfkcJCwDwnZn6z3wXmqPV" } ], "service": [ { "id": "did:ethr:0x5a2e...#vcs", "type": "VerifiableCredentialService", "serviceEndpoint": "https://example.com/vcs/" } ] } DID方法实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 // Ethereum DID Registry import { ethers } from 'ethers' class EthrDID { private registry: ethers.Contract private provider: ethers.Provider constructor() { this.provider = new ethers.JsonRpcProvider('https://eth.llamarpc.com') this.registry = new ethers.Contract( '0xdca7ef03e98e0dc2b855be647c39abe984fcf21b', ['function owner(address) view returns (address)'], this.provider ) } async resolve(did: string): Promise<DIDDocument> { // 解析DID const [, , address] = did.split(':') // 从链上获取DID文档 const owner = await this.registry.owner(address) return { '@context': 'https://www.w3.org/ns/did/v1', id: did, verificationMethod: [{ id: `${did}#controller`, type: 'EcdsaSecp256k1RecoveryMethod2020', controller: did, blockchainAccountId: `${address}@eip155:1` }], authentication: [`${did}#controller`], assertionMethod: [`${did}#controller`] } } async createDID(privateKey: string): Promise<string> { const wallet = new ethers.Wallet(privateKey) const address = await wallet.getAddress() return `did:ethr:${address}` } } // 使用 const ethrDid = new EthrDID() const did = await ethrDid.createDID(privateKey) // did:ethr:0x5a2e... const document = await ethrDid.resolve(did) 可验证凭证（Verifiable Credentials） VC数据模型 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 interface VerifiableCredential { '@context': string[] | string type: string[] id?: string issuer: string | Issuer issuanceDate: string expirationDate?: string credentialSubject: CredentialSubject credentialStatus?: CredentialStatus refreshService?: RefreshService termsOfUse?: TermsOfUse[] evidence?: Evidence[] proof?: Proof } // 示例：大学学历凭证 const universityDegree: VerifiableCredential = { '@context': [ 'https://www.w3.org/2018/credentials/v1', 'https://www.w3.org/2018/credentials/examples/v1' ], type: ['VerifiableCredential', 'UniversityDegreeCredential'], id: 'urn:uuid:12345678-1234-5678-1234-567812345678', issuer: 'did:ethr:0x1234...', issuanceDate: '2024-01-06T12:00:00Z', expirationDate: '2034-01-06T12:00:00Z', credentialSubject: { id: 'did:ethr:0xabcd...', degree: { type: 'BachelorDegree', name: '计算机科学学士' }, university: '示例大学' }, proof: { type: 'EcdsaSecp256k1Signature2019', created: '2024-01-06T12:00:00Z', proofPurpose: 'assertionMethod', verificationMethod: 'did:ethr:0x1234...#controller', jws: 'eyJhbGciOiJFUzI1Nk...' // 签名 } } VC发行 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 import { ethers } from 'ethers' import { createVerifiableCredentialJwt } from 'did-jwt-vc' class CredentialIssuer { private issuerDid: string private issuerWallet: ethers.Wallet constructor(did: string, privateKey: string) { this.issuerDid = did this.issuerWallet = new ethers.Wallet(privateKey) } async issueCredential( subjectDid: string, claims: object, expiresIn: string = '1y' ): Promise<string> { const vc: VerifiableCredential = { '@context': ['https://www.w3.org/2018/credentials/v1'], type: ['VerifiableCredential'], issuer: this.issuerDid, issuanceDate: new Date().toISOString(), expirationDate: new Date(Date.now() + this.parseExpiration(expiresIn)).toISOString(), credentialSubject: { id: subjectDid, ...claims } } // 创建签名 const signer = this.issuerWallet.signMessage.bind(this.issuerWallet) // 生成JWT格式的VC const vcJwt = await createVerifiableCredentialJwt( vc, { issuer: this.issuerDid, signer } ) return vcJwt } parseExpiration(expiresIn: string): number { const match = expiresIn.match(/^(\d+)([dhmy])$/) if (!match) throw new Error('Invalid expiration format') const value = parseInt(match[1]) const unit = match[2] const multipliers = { 'd': 86400000, 'h': 3600000, 'm': 60000, 'y': 31536000000 } return value * multipliers[unit] } } // 使用示例 const issuer = new CredentialIssuer( 'did:ethr:0x1234...', '0x私钥' ) const vcJwt = await issuer.issueCredential( 'did:ethr:0xabcd...', { degree: { type: 'BachelorDegree', name: '计算机科学学士' }, university: '示例大学' }, '1y' ) VC验证 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 import { verifyCredential, verifyPresentation } from 'did-jwt-vc' import { resolveDid } from '@identitybuilding/did-resolver' class CredentialVerifier { private didResolver: any constructor() { this.didResolver = resolveDid } async verifyCredential(vcJwt: string): Promise<VerificationResult> { try { // 验证签名 const verifiedVC = await verifyCredential(vcJwt, { resolver: this.didResolver }) // 检查过期 if (verifiedVC.expirationDate) { const expirationDate = new Date(verifiedVC.expirationDate) if (expirationDate < new Date()) { return { valid: false, reason: 'Credential has expired' } } } // 检查撤销状态 const status = await this.checkRevocation(verifiedVC) if (!status.valid) { return { valid: false, reason: 'Credential has been revoked' } } return { valid: true, credential: verifiedVC } } catch (error) { return { valid: false, reason: error.message } } } async checkRevocation(vc: VerifiableCredential): Promise<{ valid: boolean }> { if (!vc.credentialStatus) { return { valid: true } } const { id, type } = vc.credentialStatus if (type === 'RevocationList2021') { // 检查比特映射 const index = parseInt(id.split('#')[1]) const revoked = await this.checkBitMap(index) return { valid: !revoked } } return { valid: true } } async checkBitMap(index: number): Promise<boolean> { // 从链上或IPFS获取撤销列表 // ... return false } } // 使用示例 const verifier = new CredentialVerifier() const result = await verifier.verifyCredential(vcJwt) if (result.valid) { console.log('凭证有效', result.credential) } else { console.log('凭证无效:', result.reason) } VP（可验证表达） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 // Verifiable Presentation interface VerifiablePresentation { '@context': string[] | string type: string[] id?: string verifiableCredential?: VerifiableCredential[] holder?: string proof?: Proof } // 创建VP import { createVerifiablePresentationJwt } from 'did-jwt-vc' class PresentationHolder { private holderDid: string private holderWallet: ethers.Wallet constructor(did: string, privateKey: string) { this.holderDid = did this.holderWallet = new ethers.Wallet(privateKey) } async createPresentation( vcs: string[], audience: string ): Promise<string> { const vp: VerifiablePresentation = { '@context': ['https://www.w3.org/2018/credentials/v1'], type: ['VerifiablePresentation'], holder: this.holderDid, verifiableCredential: vcs.map(vcJwt => { // 简化版，实际应该解析JWT return { '@context': '...', type: ['VerifiableCredential'], ... } }) } const signer = this.holderWallet.signMessage.bind(this.holderWallet) const vpJwt = await createVerifiablePresentationJwt( vp, { audience, holder: this.holderDid, signer } ) return vpJwt } } // 使用 const holder = new PresentationHolder( 'did:ethr:0xabcd...', '0x用户私钥' ) const vpJwt = await holder.createPresentation( [vcJwt], 'did:ethr:0x9999...' // verifier的DID ) 链上身份协议 Lens Protocol 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 // Lens Profile NFT interface LensProfile { handle: string // @username imageURI: string // 头像 followModule: FollowModule followNFTURI: string dispatcher: Address } // 创建Lens Profile import { providers, Contract, utils } from 'ethers' class LensProfileManager { private lensHub: Contract private provider: providers.Provider constructor(rpcUrl: string) { this.provider = new providers.JsonRpcProvider(rpcUrl) this.lensHub = new Contract( '0xDb46d1Dc155634FfC7D94Fda11Bc2b0D29Ad869d', // LensHub on Polygon [ 'function createProfile(tuple(tuple(string handle,string imageURI) profile)) external', 'function getDefaultProfile(address) view returns (uint256)', 'function getProfile(uint256) view returns (tuple(...))' ], new ethers.Wallet(process.env.PRIVATE_KEY, this.provider) ) } async createProfile(handle: string, imageURI: string): Promise<string> { const tx = await this.lensHub.createProfile({ profile: { handle, imageURI } }) const receipt = await tx.wait() console.log('Profile created:', receipt.transactionHash) return receipt.transactionHash } async getProfile(profileId: number): Promise<LensProfile> { const profile = await this.lensHub.getProfile(profileId) return { handle: profile.handle, imageURI: profile.imageURI, followModule: profile.followModule, followNFTURI: profile.followNFTURI, dispatcher: profile.dispatcher } } } // 使用 const lens = new LensProfileManager('https://polygon-rpc.com') await lens.createProfile( 'myusername', 'ipfs://Qm...' ) ENS (Ethereum Name Service) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 import { providers, Contract } from 'ethers' class ENSManager { private ensRegistry: Contract private resolverContract: Contract private provider: providers.Provider constructor(rpcUrl: string) { this.provider = new providers.JsonRpcProvider(rpcUrl) // ENS Registry this.ensRegistry = new Contract( '0x00000000000C2E074eC69A0dFb2997BA6C7d2e1e', ['function owner(bytes32 node) view returns (address)'], this.provider ) // Public Resolver this.resolverContract = new Contract( '0x4976fb03C32e5B8cfe2b6cCCb85c41a121551E2F', [ 'function addr(bytes32 node) view returns (address)', 'function setText(bytes32 node, string key, string value)', 'function text(bytes32 node, string key) view returns (string)' ], this.provider ) } namehash(name: string): string { // ENS namehash算法 const node = '0x0000000000000000000000000000000000000000000000000000000000000000' if (name === '') { return node } const labels = name.split('.') let hash = node for (let i = labels.length - 1; i >= 0; i--) { const labelHash = ethers.utils.keccak256(ethers.utils.toUtf8Bytes(labels[i])) hash = ethers.utils.keccak256( ethers.utils.concat([hash, labelHash]) ) } return hash } async getAddress(name: string): Promise<string> { const node = this.namehash(name) return await this.resolverContract.addr(node) } async setText(name: string, key: string, value: string, signer: ethers.Signer) { const node = this.namehash(name) const resolverWithSigner = this.resolverContract.connect(signer) const tx = await resolverWithSigner.setText(node, key, value) await tx.wait() } async getText(name: string, key: string): Promise<string> { const node = this.namehash(name) return await this.resolverContract.text(node, key) } } // 使用 const ens = new ENSManager('https://eth.llamarpc.com') // 解析ENS域名 const address = await ens.getAddress('vitalik.eth') // 0xab5801a7D398351b8bE11C439e05C5B3259aEbC4 // 设置和读取ENS记录 await ens.setText( 'mydomain.eth', 'com.twitter', '@myhandle', signer ) Soulbound Token (SBT) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 // SBT（灵魂绑定代币）是不可转移的NFT import { ethers } from 'ethers' // SBT合约ABI const SBT_ABI = [ 'function issue(address to, uint256 tokenId, string uri) external', 'function revoke(address from, uint256 tokenId) external', 'function tokenURI(uint256 tokenId) view returns (string)', 'function balanceOf(address) view returns (uint256)', 'function tokenOfOwnerByIndex(address owner, uint256 index) view returns (uint256)' ] class SoulboundManager { private sbtContract: ethers.Contract constructor(contractAddress: string, privateKey: string) { const provider = new ethers.JsonRpcProvider('https://eth.llamarpc.com') const wallet = new ethers.Wallet(privateKey, provider) this.sbtContract = new ethers.Contract( contractAddress, SBT_ABI, wallet ) } async issueSBT( recipient: string, tokenId: number, metadataURI: string ): Promise<string> { const tx = await this.sbtContract.issue( recipient, tokenId, metadataURI ) const receipt = await tx.wait() return receipt.transactionHash } async getSBTsByAddress(address: string): Promise<number[]> { const balance = await this.sbtContract.balanceOf(address) const tokens = [] for (let i = 0; i < balance.toNumber(); i++) { const tokenId = await this.sbtContract.tokenOfOwnerByIndex(address, i) tokens.push(tokenId.toNumber()) } return tokens } async getSBTMetadata(tokenId: number): Promise<object> { const uri = await this.sbtContract.tokenURI(tokenId) // 从IPFS获取metadata const response = await fetch(uri) const metadata = await response.json() return metadata } } // 使用示例 const sbtManager = new SoulboundManager( '0x...', // SBT合约地址 process.env.PRIVATE_KEY ) // 发行SBT凭证 await sbtManager.issueSBT( '0x用户地址', 1, 'ipfs://Qm...' // metadata URI ) // 查询用户的SBT const tokens = await sbtManager.getSBTsByAddress('0x用户地址') for (const tokenId of tokens) { const metadata = await sbtManager.getSBTMetadata(tokenId) console.log('SBT:', metadata) // { // name: "大学学历凭证", // description: "计算机科学学士学位", // image: "ipfs://...", // attributes: [ // { trait_type: "大学", value: "示例大学" }, // { trait_type: "专业", value: "计算机科学" }, // { trait_type: "学位", value: "学士" } // ] // } } 链上声誉系统 信任分数算法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 interface ReputationData { totalInteractions: number successfulInteractions: number averageRating: number stakingAmount: number accountAge: number } class ReputationCalculator { calculateScore(data: ReputationData): number { let score = 50 // 基础分 // 交互成功率（+30分） if (data.totalInteractions > 0) { const successRate = data.successfulInteractions / data.totalInteractions score += successRate * 30 } // 平均评分（+10分） score += (data.averageRating - 3) * 3.33 // 1-5分制 // 质押金额（+5分） const stakingBonus = Math.log10(data.stakingAmount + 1) * 2 score += Math.min(stakingBonus, 5) // 账号年龄（+5分） const ageInYears = data.accountAge / (365 * 24 * 60 * 60) score += Math.min(ageInYears * 2, 5) return Math.min(Math.max(score, 0), 100) } calculateTier(score: number): string { if (score >= 90) return '钻石级' if (score >= 75) return '黄金级' if (score >= 60) return '白银级' if (score >= 40) return '青铜级' return '新手级' } } // 链上声誉合约 import { ethers } from 'ethers' class OnChainReputation { private contract: ethers.Contract constructor() { const provider = new ethers.JsonRpcProvider('https://polygon-rpc.com') const wallet = new ethers.Wallet(process.env.PRIVATE_KEY, provider) this.contract = new ethers.Contract( '0x...', // 声誉合约地址 [ 'function recordInteraction(address user, bool success, uint8 rating) external', 'function getReputationScore(address user) view returns (uint256)', 'function stakeTokens(uint256 amount) external', 'function unstakeTokens(uint256 amount) external', 'event ReputationUpdated(address indexed user, uint256 score)' ], wallet ) } async recordInteraction( user: string, success: boolean, rating: number ): Promise<void> { const tx = await this.contract.recordInteraction( user, success, rating ) await tx.wait() console.log('Interaction recorded') } async getReputationScore(user: string): Promise<number> { const score = await this.contract.getReputationScore(user) return score.toNumber() } async stakeTokens(amount: number): Promise<void> { const tx = await this.contract.stakeTokens( ethers.utils.parseEther(amount.toString()) ) await tx.wait() console.log('Tokens staked') } listenToReputationUpdates() { this.contract.on('ReputationUpdated', (user, score) => { console.log(`Reputation updated for ${user}: ${score}`) }) } } DID钱包实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 import { ethers } from 'ethers' import { hdkey } from 'ethereumjs-wallet' import * as bip39 from 'bip39' class DIDWallet { private mnemonic: string private hdNode: any private did: string private credentials: string[] = [] constructor() { this.generate() } private generate() { // 生成助记词 this.mnemonic = bip39.generateMnemonic() // 从助记词生成HD钱包 const seed = bip39.mnemonicToSeedSync(this.mnemonic) this.hdNode = hdkey.fromMasterSeed(seed) // 派生第一个账户作为DID const path = "m/44'/60'/0'/0/0" const wallet = this.hdNode.derivePath(path).getWallet() const address = wallet.getAddressString() this.did = `did:ethr:${address}` } getDID(): string { return this.did } getAddress(): string { return this.did.split(':')[2] } async sign(message: string): Promise<string> { const path = "m/44'/60'/0'/0/0" const wallet = this.hdNode.derivePath(path).getWallet() const signature = await wallet.signMessage(message) return signature } async verify(message: string, signature: string): Promise<boolean> { const address = ethers.utils.verifyMessage(message, signature) return address.toLowerCase() === this.getAddress().toLowerCase() } addCredential(vcJwt: string) { this.credentials.push(vcJwt) } getCredentials(): string[] { return this.credentials } exportWallet(): string { return JSON.stringify({ mnemonic: this.mnemonic, did: this.did, credentials: this.credentials }) } importWallet(data: string) { const wallet = JSON.parse(data) this.mnemonic = wallet.mnemonic this.did = wallet.did this.credentials = wallet.credentials || [] const seed = bip39.mnemonicToSeedSync(this.mnemonic) this.hdNode = hdkey.fromMasterSeed(seed) } } // React Hook import { useState, useEffect } from 'react' export function useDIDWallet() { const [wallet, setWallet] = useState<DIDWallet | null>(null) useEffect(() => { // 从localStorage加载钱包 const savedWallet = localStorage.getItem('did-wallet') if (savedWallet) { const newWallet = new DIDWallet() newWallet.importWallet(savedWallet) setWallet(newWallet) } else { // 创建新钱包 const newWallet = new DIDWallet() setWallet(newWallet) // 保存到localStorage localStorage.setItem('did-wallet', newWallet.exportWallet()) } }, []) const backupWallet = () => { if (wallet) { const data = wallet.exportWallet() const blob = new Blob([data], { type: 'application/json' }) const url = URL.createObjectURL(blob) const a = document.createElement('a') a.href = url a.download = `did-wallet-${Date.now()}.json` a.click() } } return { wallet, backupWallet } } 完整DID应用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 import { useState } from 'react' import { DIDWallet } from './did-wallet' import { CredentialIssuer } from './credential-issuer' import { CredentialVerifier } from './credential-verifier' export default function DIDApp() { const [wallet] = useState(() => new DIDWallet()) const [credentials, setCredentials] = useState<string[]>([]) const [presentation, setPresentation] = useState<string>('') const issueCredential = async (claims: object) => { const issuer = new CredentialIssuer( wallet.getDID(), '0x...' // 发行方私钥 ) const vcJwt = await issuer.issueCredential( wallet.getDID(), claims ) setCredentials([...credentials, vcJwt]) } const createPresentation = async () => { const holder = new PresentationHolder( wallet.getDID(), '0x...' // 用户私钥 ) const vpJwt = await holder.createPresentation( credentials, 'did:ethr:0x9999...' // verifier DID ) setPresentation(vpJwt) } const verifyPresentation = async (vpJwt: string) => { const verifier = new CredentialVerifier() const result = await verifier.verifyPresentation(vpJwt) return result } return ( <div> <h1>DID身份钱包</h1> <div> <h2>我的DID</h2> <p>{wallet.getDID()}</p> </div> <div> <h2>凭证</h2> {credentials.map((vc, index) => ( <div key={index}> <pre>{vc}</pre> </div> ))} </div> <div> <h2>可验证表达</h2> <pre>{presentation}</pre> </div> <button onClick={() => issueCredential({ name: '张三' })}> 发行凭证 </button> <button onClick={createPresentation}> 创建表达 </button> </div> ) } 总结 去中心化身份（DID）是WEB3的重要基础设施，通过W3C标准化、区块链技术和密码学，为用户提供了真正自主可控的身份系统。 ...

引言 RAG（Retrieval-Augmented Generation，检索增强生成）技术是大模型应用的核心范式之一。它通过检索外部知识库来增强生成能力，有效解决了大模型知识滞后、幻觉严重等问题。本文将深入探讨RAG技术的完整实现链路，从文档处理到检索生成，帮助开发者构建企业级知识问答系统。 RAG技术概述 为什么需要RAG 1 2 3 4 5 6 7 8 9 10 11 12 # 纯LLM生成的问题 response = llm.generate("公司2024年Q3财报数据是多少？") # 问题1：模型知识截止，不知道最新信息 # 问题2：可能产生幻觉，编造数据 # RAG方案 relevant_docs = retriever.search("2024年Q3财报") response = llm.generate( prompt=f"基于以下文档回答问题：\n{relevant_docs}\n\n问题：公司2024年Q3财报数据是多少？" ) # 优势1：基于真实文档，准确可靠 # 优势2：可利用最新知识，实时更新 RAG核心流程 文档摄入 → 文本分块 → 向量化 → 向量数据库 ↓ 查询 → 向量化 → 检索 → 重排序 → LLM生成 → 答案 文档处理 1. 文档解析 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 from typing import List, Dict import pypdf from docx import Document import markdown class DocumentParser: """多格式文档解析器""" def parse(self, file_path: str) -> List[Dict]: """解析文档""" ext = self.get_extension(file_path) if ext == ".pdf": return self.parse_pdf(file_path) elif ext == ".docx": return self.parse_docx(file_path) elif ext in [".md", ".markdown"]: return self.parse_markdown(file_path) elif ext == ".txt": return self.parse_text(file_path) else: raise ValueError(f"Unsupported format: {ext}") def parse_pdf(self, file_path: str) -> List[Dict]: """解析PDF文件""" pages = [] with open(file_path, 'rb') as file: pdf_reader = pypdf.PdfReader(file) for page_num, page in enumerate(pdf_reader.pages): text = page.extract_text() # 提取元数据 metadata = { "page": page_num + 1, "source": file_path, "total_pages": len(pdf_reader.pages) } pages.append({ "content": text, "metadata": metadata }) return pages def parse_docx(self, file_path: str) -> List[Dict]: """解析Word文档""" doc = Document(file_path) paragraphs = [] for para_num, para in enumerate(doc.paragraphs): if para.text.strip(): paragraphs.append({ "content": para.text, "metadata": { "paragraph": para_num, "source": file_path } }) return paragraphs def parse_markdown(self, file_path: str) -> List[Dict]: """解析Markdown文件""" with open(file_path, 'r', encoding='utf-8') as f: md_content = f.read() # 使用markdown解析器 md = markdown.Markdown() html = md.convert(md_content) # 提取标题层级结构 sections = self.extract_sections(html, md_content) return sections def extract_sections(self, html: str, md: str) -> List[Dict]: """提取Markdown章节""" from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') sections = [] current_section = {"headers": [], "content": []} for element in soup.find_all(['h1', 'h2', 'h3', 'p', 'code']): if element.name.startswith('h'): # 保存之前的section if current_section["content"]: sections.append({ "content": "\n".join(current_section["content"]), "metadata": { "headers": current_section["headers"] } }) # 开始新的section level = int(element.name[1]) current_section = { "headers": current_section["headers"][:level-1] + [element.text], "content": [] } else: current_section["content"].append(element.get_text()) # 添加最后一个section if current_section["content"]: sections.append({ "content": "\n".join(current_section["content"]), "metadata": { "headers": current_section["headers"] } }) return sections 2. 文本分块 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 from typing import List import re from sentence_transformers import SentenceTransformer class TextChunker: """文本分块器""" def __init__(self, method: str = "recursive"): self.method = method self.embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') def chunk(self, text: str, **kwargs) -> List[str]: """分块""" if self.method == "fixed_size": return self.fixed_size_chunk(text, **kwargs) elif self.method == "recursive": return self.recursive_chunk(text, **kwargs) elif self.method == "semantic": return self.semantic_chunk(text, **kwargs) else: raise ValueError(f"Unknown chunking method: {self.method}") def fixed_size_chunk( self, text: str, chunk_size: int = 500, overlap: int = 50 ) -> List[str]: """固定大小分块""" chunks = [] start = 0 text_length = len(text) while start < text_length: end = start + chunk_size chunk = text[start:end] chunks.append(chunk) # 移动到下一个块（考虑重叠） start = end - overlap return chunks def recursive_chunk( self, text: str, separators: List[str] = None, chunk_size: int = 1000, overlap: int = 100 ) -> List[str]: """递归分块 - 按分隔符智能分块""" if separators is None: separators = ["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""] # 尝试按分隔符分割 for separator in separators: if separator in text: parts = text.split(separator) chunks = [] current_chunk = "" for part in parts: # 如果添加这个部分会超过限制 if len(current_chunk) + len(part) + len(separator) > chunk_size: if current_chunk: chunks.append(current_chunk) # 如果单个部分本身就太长，递归处理 if len(part) > chunk_size: sub_chunks = self.recursive_chunk( part, separators[separators.index(separator) + 1:], chunk_size, overlap ) chunks.extend(sub_chunks) current_chunk = "" else: current_chunk = part + separator else: current_chunk += part + separator # 添加最后一个块 if current_chunk: chunks.append(current_chunk) return chunks # 如果没有找到分隔符，使用固定大小分块 return self.fixed_size_chunk(text, chunk_size, overlap) def semantic_chunk( self, text: str, max_similarity: float = 0.7 ) -> List[str]: """语义分块 - 基于语义相似度""" # 首先按句子分割 sentences = re.split(r'([。！？.!?])', text) sentences = [ s1 + s2 for s1, s2 in zip(sentences[::2], sentences[1::2]) if s1.strip() ] if not sentences: return [text] chunks = [] current_chunk = [sentences[0]] for sentence in sentences[1:]: # 计算与当前块最后一个句子的相似度 last_sentence = current_chunk[-1] similarity = self.compute_similarity(last_sentence, sentence) # 如果相似度高，合并到当前块 if similarity < max_similarity: current_chunk.append(sentence) else: # 否则开始新块 chunks.append("".join(current_chunk)) current_chunk = [sentence] # 添加最后一个块 if current_chunk: chunks.append("".join(current_chunk)) return chunks def compute_similarity(self, text1: str, text2: str) -> float: """计算语义相似度""" emb1 = self.embedding_model.encode(text1) emb2 = self.embedding_model.encode(text2) # 余弦相似度 import numpy as np return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)) 3. 元数据提取 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 class MetadataExtractor: """元数据提取器""" def __init__(self): self.date_patterns = [ r'\d{4}-\d{2}-\d{2}', r'\d{4}年\d{1,2}月\d{1,2}日', r'\d{4}/\d{2}/\d{2}' ] self.url_pattern = r'https?://[^\s]+' self.email_pattern = r'\w+@\w+\.\w+' def extract(self, chunk: str) -> Dict: """提取元数据""" metadata = {} # 提取日期 metadata['dates'] = self.extract_dates(chunk) # 提取URL metadata['urls'] = self.extract_urls(chunk) # 提取邮箱 metadata['emails'] = self.extract_emails(chunk) # 提取关键词 metadata['keywords'] = self.extract_keywords(chunk) # 提取实体 metadata['entities'] = self.extract_entities(chunk) return metadata def extract_dates(self, text: str) -> List[str]: """提取日期""" dates = [] for pattern in self.date_patterns: dates.extend(re.findall(pattern, text)) return dates def extract_urls(self, text: str) -> List[str]: """提取URL""" return re.findall(self.url_pattern, text) def extract_emails(self, text: str) -> List[str]: """提取邮箱""" return re.findall(self.email_pattern, text) def extract_keywords(self, text: str, top_k: int = 5) -> List[str]: """提取关键词""" # 使用TF-IDF或RAKE算法 from sklearn.feature_extraction.text import TfidfVectorizer vectorizer = TfidfVectorizer(max_features=top_k) tfidf_matrix = vectorizer.fit_transform([text]) feature_names = vectorizer.get_feature_names_out() tfidf_scores = tfidf_matrix.toarray()[0] # 获取top-k关键词 top_indices = tfidf_scores.argsort()[-top_k:][::-1] return [feature_names[i] for i in top_indices] def extract_entities(self, text: str) -> Dict[str, List[str]]: """提取命名实体""" # 使用spaCy或其他NER工具 import spacy nlp = spacy.load("zh_core_web_sm") doc = nlp(text) entities = { "PERSON": [], "ORG": [], "GPE": [], "DATE": [] } for ent in doc.ents: if ent.label_ in entities: entities[ent.label_].append(ent.text) return entities 向量数据库 1. ChromaDB集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 import chromadb from chromadb.config import Settings class ChromaVectorStore: """ChromaDB向量存储""" def __init__(self, collection_name: str = "documents"): self.client = chromadb.Client(Settings( chroma_db_impl="duckdb+parquet", persist_directory="./chroma_db" )) # 创建或获取collection self.collection = self.client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} ) def add_documents( self, documents: List[str], embeddings: List[List[float]], metadatas: List[Dict], ids: List[str] ): """添加文档""" self.collection.add( documents=documents, embeddings=embeddings, metadatas=metadatas, ids=ids ) def search( self, query_embedding: List[float], top_k: int = 5, where: Dict = None ) -> Dict: """搜索相似文档""" results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k, where=where ) return { "documents": results["documents"][0], "metadatas": results["metadatas"][0], "distances": results["distances"][0] } def delete(self, ids: List[str]): """删除文档""" self.collection.delete(ids=ids) def update( self, ids: List[str], documents: List[str] = None, embeddings: List[List[float]] = None, metadatas: List[Dict] = None ): """更新文档""" self.collection.update( ids=ids, documents=documents, embeddings=embeddings, metadatas=metadatas ) 2. Pinecone集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 import pinecone from pinecone import ServerlessSpec class PineconeVectorStore: """Pinecone向量存储""" def __init__(self, api_key: str, environment: str): pinecone.init(api_key=api_key, environment=environment) self.index_name = "rag-index" self.dimension = 768 # 根据embedding模型调整 # 创建index如果不存在 if self.index_name not in pinecone.list_indexes(): pinecone.create_index( name=self.index_name, dimension=self.dimension, metric="cosine", spec=ServerlessSpec( cloud="aws", region="us-east-1" ) ) self.index = pinecone.Index(self.index_name) def upsert(self, vectors: List[Dict]): """批量插入/更新向量""" self.index.upsert(vectors=vectors) def query( self, vector: List[float], top_k: int = 5, filter: Dict = None, include_metadata: bool = True ) -> Dict: """查询相似向量""" return self.index.query( vector=vector, top_k=top_k, filter=filter, include_metadata=include_metadata ) def delete(self, ids: List[str]): """删除向量""" self.index.delete(ids=ids) 3. Milvus集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType class MilvusVectorStore: """Milvus向量存储""" def __init__(self, host: str = "localhost", port: int = 19530): # 连接Milvus connections.connect("default", host=host, port=port) # 定义collection schema self.collection_name = "rag_documents" self.dimension = 768 fields = [ FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=100), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dimension), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="metadata", dtype=DataType.JSON) ] schema = CollectionSchema(fields, f"{self.collection_name} schema") # 创建collection if self.collection_name not in [c.name for c in connections.list_connections()]: self.collection = Collection( name=self.collection_name, schema=schema ) else: self.collection = Collection(self.collection_name) # 创建索引 index_params = { "index_type": "IVF_FLAT", "metric_type": "COSINE", "params": {"nlist": 128} } self.collection.create_index( field_name="embedding", index_params=index_params ) self.collection.load() def insert(self, data: List[Dict]): """插入数据""" self.collection.insert(data) def search( self, embedding: List[float], top_k: int = 5, expr: str = None ) -> Dict: """搜索""" results = self.collection.search( data=[embedding], anns_field="embedding", param={"metric_type": "COSINE", "params": {"nprobe": 10}}, limit=top_k, expr=expr, output_fields=["content", "metadata"] ) return results[0] 检索策略 1. 语义检索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 from sentence_transformers import SentenceTransformer import numpy as np class SemanticRetriever: """语义检索器""" def __init__( self, vector_store: ChromaVectorStore, model_name: str = "paraphrase-multilingual-MiniLM-L12-v2" ): self.vector_store = vector_store self.embedding_model = SentenceTransformer(model_name) def retrieve( self, query: str, top_k: int = 5, filters: Dict = None ) -> List[Dict]: """检索相关文档""" # 生成查询向量 query_embedding = self.embedding_model.encode(query) # 向量搜索 results = self.vector_store.search( query_embedding=query_embedding.tolist(), top_k=top_k, where=filters ) # 格式化结果 documents = [] for i, doc in enumerate(results["documents"]): documents.append({ "content": doc, "metadata": results["metadatas"][i], "score": 1 - results["distances"][i] # 转换为相似度 }) return documents 2. 混合检索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 class HybridRetriever: """混合检索器（语义+关键词）""" def __init__( self, vector_store: ChromaVectorStore, keyword_index: object, # Elasticsearch或Whoosh semantic_weight: float = 0.7 ): self.vector_store = vector_store self.keyword_index = keyword_index self.semantic_weight = semantic_weight def retrieve( self, query: str, top_k: int = 5 ) -> List[Dict]: """混合检索""" # 语义检索 semantic_results = self.semantic_search(query, top_k * 2) # 关键词检索 keyword_results = self.keyword_search(query, top_k * 2) # 归一化分数 semantic_scores = self.normalize_scores([r["score"] for r in semantic_results]) keyword_scores = self.normalize_scores([r["score"] for r in keyword_results]) # 合并分数 combined_scores = {} for doc, score in zip(semantic_results, semantic_scores): doc_id = doc["metadata"]["id"] combined_scores[doc_id] = score * self.semantic_weight for doc, score in zip(keyword_results, keyword_scores): doc_id = doc["metadata"]["id"] if doc_id in combined_scores: combined_scores[doc_id] += score * (1 - self.semantic_weight) else: combined_scores[doc_id] = score * (1 - self.semantic_weight) # 排序并返回top-k sorted_docs = sorted( combined_scores.items(), key=lambda x: x[1], reverse=True )[:top_k] return [self.get_document(doc_id) for doc_id, _ in sorted_docs] def normalize_scores(self, scores: List[float]) -> List[float]: """归一化分数到0-1""" min_score = min(scores) max_score = max(scores) if max_score == min_score: return [1.0] * len(scores) return [ (score - min_score) / (max_score - min_score) for score in scores ] 3. 重排序 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 from sentence_transformers import CrossEncoder class Reranker: """重排序器""" def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"): self.reranker = CrossEncoder(model_name) def rerank( self, query: str, documents: List[Dict], top_k: int = 5 ) -> List[Dict]: """重新排序""" # 准备query-document对 pairs = [ (query, doc["content"]) for doc in documents ] # 计算重排序分数 scores = self.reranker.predict(pairs) # 添加分数到文档 for doc, score in zip(documents, scores): doc["rerank_score"] = float(score) # 按重排序分数排序 documents.sort(key=lambda x: x["rerank_score"], reverse=True) return documents[:top_k] 生成增强 1. Prompt构建 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 class PromptBuilder: """Prompt构建器""" def __init__(self, template_type: str = "qa"): self.template_type = template_type def build( self, query: str, context: List[Dict], chat_history: List[Dict] = None ) -> str: """构建prompt""" if self.template_type == "qa": return self.build_qa_prompt(query, context) elif self.template_type == "conversational": return self.build_conversational_prompt(query, context, chat_history) elif self.template_type == "structured": return self.build_structured_prompt(query, context) else: raise ValueError(f"Unknown template type: {self.template_type}") def build_qa_prompt(self, query: str, context: List[Dict]) -> str: """构建QA prompt""" # 格式化上下文 context_text = "\n\n".join([ f"文档{i+1}:\n{doc['content']}\n来源: {doc['metadata'].get('source', 'unknown')}" for i, doc in enumerate(context) ]) prompt = f""" 请基于以下文档内容回答问题。如果文档中没有相关信息，请明确说明。 ### 参考文档 {context_text} ### 问题 {query} ### 回答要求 1. 只使用参考文档中的信息 2. 如果文档中没有答案，明确说明"根据提供的文档，我无法回答这个问题" 3. 引用具体来源 4. 保持准确和客观 回答： """ return prompt.strip() def build_conversational_prompt( self, query: str, context: List[Dict], chat_history: List[Dict] ) -> str: """构建对话式prompt""" context_text = "\n\n".join([ f"- {doc['content']}" for doc in context ]) # 格式化历史对话 history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in chat_history[-5:] # 只保留最近5轮 ]) prompt = f""" ### 参考信息 {context_text} ### 对话历史 {history_text} ### 当前问题 {query} ### 要求 1. 基于参考信息回答 2. 考虑对话历史上下文 3. 保持自然对话风格 4. 如果参考信息不足，可以基于常识补充 回答： """ return prompt.strip() def build_structured_prompt(self, query: str, context: List[Dict]) -> str: """构建结构化prompt（用于生成结构化输出）""" context_text = "\n\n".join([ f"【{doc['metadata'].get('title', '文档')}】\n{doc['content']}" for doc in context ]) prompt = f""" ## 任务说明 请基于以下参考文档回答用户问题，并按指定格式输出。 ## 参考文档 {context_text} ## 用户问题 {query} ## 输出格式要求 请按以下JSON格式输出： {{ "answer": "详细回答", "confidence": "高/中/低", "sources": ["来源1", "来源2"], "key_points": ["要点1", "要点2"] }} 请输出： """ return prompt.strip() 2. LLM生成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 import openai from typing import Dict, List class RAGGenerator: """RAG生成器""" def __init__( self, api_key: str, model: str = "gpt-4", temperature: float = 0.7 ): openai.api_key = api_key self.model = model self.temperature = temperature def generate( self, query: str, context: List[Dict], stream: bool = False ) -> Dict: """生成回答""" # 构建prompt prompt_builder = PromptBuilder(template_type="qa") prompt = prompt_builder.build(query, context) # 调用LLM response = openai.ChatCompletion.create( model=self.model, messages=[ { "role": "system", "content": "你是一个专业的知识助手，擅长基于提供的文档回答问题。" }, { "role": "user", "content": prompt } ], temperature=self.temperature, stream=stream ) if stream: # 流式输出 return self._stream_response(response) else: # 一次性返回 answer = response.choices[0].message.content return { "answer": answer, "sources": [doc["metadata"] for doc in context], "model": self.model } def _stream_response(self, response): """处理流式响应""" for chunk in response: if chunk.choices[0].delta.get("content"): yield chunk.choices[0].delta.content 完整RAG系统 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 class RAGSystem: """完整RAG系统""" def __init__(self, config: Dict): # 初始化组件 self.embedder = SentenceTransformer(config["embedding_model"]) self.vector_store = ChromaVectorStore(config["collection_name"]) self.retriever = SemanticRetriever(self.vector_store) self.reranker = Reranker(config["reranker_model"]) self.generator = RAGGenerator(config["openai_api_key"]) self.prompt_builder = PromptBuilder(template_type="qa") # 文档处理 self.parser = DocumentParser() self.chunker = TextChunker(method="recursive") self.metadata_extractor = MetadataExtractor() def ingest_documents(self, file_paths: List[str]): """摄取文档""" all_chunks = [] for file_path in file_paths: # 解析文档 documents = self.parser.parse(file_path) # 分块 for doc in documents: chunks = self.chunker.chunk( doc["content"], chunk_size=1000, overlap=200 ) # 提取元数据 for i, chunk in enumerate(chunks): metadata = { **doc["metadata"], "chunk_index": i, "source_file": file_path } # 额外元数据提取 extra_metadata = self.metadata_extractor.extract(chunk) metadata.update(extra_metadata) all_chunks.append({ "content": chunk, "metadata": metadata }) # 生成嵌入 embeddings = self.embedder.encode([c["content"] for c in all_chunks]) # 存储到向量数据库 ids = [f"doc_{i}" for i in range(len(all_chunks))] self.vector_store.add_documents( documents=[c["content"] for c in all_chunks], embeddings=embeddings.tolist(), metadatas=[c["metadata"] for c in all_chunks], ids=ids ) return len(all_chunks) def query( self, question: str, top_k: int = 5, rerank: bool = True ) -> Dict: """查询""" # 检索 retrieved_docs = self.retriever.retrieve( query=question, top_k=top_k * 2 # 检索更多用于重排序 ) # 重排序 if rerank: retrieved_docs = self.reranker.rerank( query=question, documents=retrieved_docs, top_k=top_k ) # 生成回答 response = self.generator.generate( query=question, context=retrieved_docs ) return response def chat( self, message: str, chat_history: List[Dict] = None, top_k: int = 3 ) -> Dict: """对话模式""" # 从历史中提取上下文 if chat_history: # 可以使用历史对话优化检索 context_query = self._build_context_query(message, chat_history) else: context_query = message # 检索 retrieved_docs = self.retriever.retrieve( query=context_query, top_k=top_k ) # 构建对话式prompt prompt_builder = PromptBuilder(template_type="conversational") prompt = prompt_builder.build(message, retrieved_docs, chat_history) # 生成 response = self.generator.generate( query=message, context=retrieved_docs ) # 添加到历史 if chat_history is None: chat_history = [] chat_history.append({"role": "user", "content": message}) chat_history.append({"role": "assistant", "content": response["answer"]}) response["chat_history"] = chat_history return response def _build_context_query(self, message: str, history: List[Dict]) -> str: """基于历史构建上下文查询""" # 提取历史中的关键词 recent_messages = history[-4:] # 最近2轮对话 context = " ".join([ msg["content"] for msg in recent_messages if msg["role"] == "user" ]) return f"{context} {message}" 优化策略 1. 查询扩展 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 class QueryExpander: """查询扩展器""" def __init__(self, llm): self.llm = llm def expand(self, query: str, num_expansions: int = 3) -> List[str]: """扩展查询""" prompt = f""" 请为以下查询生成{num_expansions}个语义相似但表述不同的查询版本。 原始查询: {query} 请只输出扩展后的查询，每行一个： """ response = self.llm.generate(prompt) expanded_queries = [query] expanded_queries.extend([ q.strip() for q in response.split('\n') if q.strip() ][:num_expansions]) return expanded_queries # 使用 expanded_queries = query_expander.expand("如何提高RAG系统性能？") # ["如何提高RAG系统性能？", "优化RAG检索效果的方法", "RAG系统性能提升技巧"] 2. Hybrid Search 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 class HybridSearchRetriever: """混合检索（稠密+稀疏）""" def retrieve(self, query: str, top_k: int = 10) -> List[Dict]: # 稠密检索（向量） dense_results = self.dense_retriever.retrieve(query, top_k * 2) # 稀疏检索（BM25） sparse_results = self.sparse_retriever.retrieve(query, top_k * 2) # RRF（Reciprocal Rank Fusion）合并 fused_results = self.rrf_fusion( dense_results, sparse_results, k=60 ) return fused_results[:top_k] def rrf_fusion( self, results1: List[Dict], results2: List[Dict], k: int = 60 ) -> List[Dict]: """RRF融合算法""" scores = {} # 计算第一个结果的分数 for rank, doc in enumerate(results1): doc_id = doc["id"] scores[doc_id] = 1 / (k + rank + 1) # 叠加第二个结果的分数 for rank, doc in enumerate(results2): doc_id = doc["id"] if doc_id in scores: scores[doc_id] += 1 / (k + rank + 1) else: scores[doc_id] = 1 / (k + rank + 1) # 排序 sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [self.get_doc(doc_id) for doc_id, _ in sorted_docs] 3. 缓存机制 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 from functools import lru_cache import hashlib class CachedRAGSystem(RAGSystem): """带缓存的RAG系统""" @lru_cache(maxsize=1000) def _cached_retrieve(self, query_hash: str, top_k: int): """缓存检索结果""" return super().retrieve(query_hash, top_k) def query(self, question: str, top_k: int = 5) -> Dict: # 生成查询hash query_hash = hashlib.md5(question.encode()).hexdigest() # 尝试从缓存获取 try: cached_result = self._cached_retrieve(query_hash, top_k) return cached_result except: # 缓存未命中，执行正常检索 result = super().query(question, top_k) return result 评估指标 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 class RAGEvaluator: """RAG系统评估器""" def __init__(self, test_data: List[Dict]): self.test_data = test_data def evaluate(self, rag_system: RAGSystem) -> Dict: """评估RAG系统""" metrics = { "retrieval_precision": [], "retrieval_recall": [], "answer_relevance": [], "faithfulness": [] } for test_case in self.test_data: question = test_case["question"] ground_truth_docs = test_case["relevant_docs"] ground_truth_answer = test_case["answer"] # 检索评估 retrieved_docs = rag_system.retriever.retrieve(question, top_k=10) retrieved_doc_ids = [doc["metadata"]["id"] for doc in retrieved_docs] precision = self.compute_precision( retrieved_doc_ids, ground_truth_docs ) recall = self.compute_recall( retrieved_doc_ids, ground_truth_docs ) metrics["retrieval_precision"].append(precision) metrics["retrieval_recall"].append(recall) # 生成评估 response = rag_system.query(question) relevance = self.compute_relevance( response["answer"], ground_truth_answer ) faithfulness = self.compute_faithfulness( response["answer"], retrieved_docs ) metrics["answer_relevance"].append(relevance) metrics["faithfulness"].append(faithfulness) # 计算平均指标 return { "avg_precision": np.mean(metrics["retrieval_precision"]), "avg_recall": np.mean(metrics["retrieval_recall"]), "avg_relevance": np.mean(metrics["answer_relevance"]), "avg_faithfulness": np.mean(metrics["faithfulness"]) } def compute_precision(self, retrieved: List, relevant: List) -> float: """计算精确率""" retrieved_set = set(retrieved) relevant_set = set(relevant) if not retrieved_set: return 0.0 return len(retrieved_set & relevant_set) / len(retrieved_set) def compute_recall(self, retrieved: List, relevant: List) -> float: """计算召回率""" retrieved_set = set(retrieved) relevant_set = set(relevant) if not relevant_set: return 0.0 return len(retrieved_set & relevant_set) / len(relevant_set) def compute_relevance(self, generated: str, reference: str) -> float: """计算答案相关性（使用余弦相似度）""" from sentence_transformers import SentenceTransformer model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2') emb1 = model.encode(generated) emb2 = model.encode(reference) import numpy as np return np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)) def compute_faithfulness(self, answer: str, contexts: List[Dict]) -> float: """计算忠实度（答案是否基于检索的上下文）""" # 使用LLM判断 prompt = f""" 请判断以下答案是否完全基于提供的上下文信息。 上下文: {chr(10).join([c['content'] for c in contexts[:3]])} 答案: {answer} 请回答"是"或"否"，并简要说明理由。 """ response = self.llm.generate(prompt) return 1.0 if "是" in response else 0.0 总结 RAG技术通过结合检索和生成，为大模型应用提供了强大的知识增强能力。本文详细介绍了从文档处理、向量存储、检索策略到生成增强的完整技术链路。 ...

引言 AI Agent（人工智能智能体）作为大语言模型最重要的应用范式之一，正在重塑我们与AI交互的方式。不同于传统的聊天机器人，AI Agent具备自主感知、决策和执行能力，能够使用工具、维护记忆、进行多步推理。本文将深入探讨AI Agent的架构设计，从理论到实践，帮助开发者构建生产级的智能体应用。 AI Agent核心概念 什么是AI Agent AI Agent是一个能够： 感知环境：理解用户输入和系统状态 推理决策：基于目标和上下文制定行动方案 执行工具：调用外部API和服务完成任务 记忆管理：维护短期和长期记忆 反思学习：从执行结果中学习和改进 Agent vs Chatbot 1 2 3 4 5 6 7 8 9 10 11 12 # 传统Chatbot chatbot_response = llm.generate("帮我查询天气") # 单轮对话，无状态，无法执行操作 # AI Agent agent = Agent( tools=[weather_api, calendar_api], memory=LongTermMemory(), planner=ReActPlanner() ) result = agent.run("帮我查明天天气，如果有雨则安排线上会议") # 多步推理，工具调用，状态管理 核心架构设计 1. 整体架构 ┌─────────────────────────────────────────────┐ │ User Interface │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Agent Core │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Planner │──│ Executor │──│ Reflector│ │ │ └──────────┘ └──────┬───┘ └──────────┘ │ │ │ │ │ ┌─────────────────────┼─────────────────┐ │ │ │ Memory System │ │ │ │ ┌─────────┐ ┌─────────┐ ┌────────┐ │ │ │ │ │ShortTerm│ │LongTerm │ │Vector │ │ │ │ │ └─────────┘ └─────────┘ └────────┘ │ │ │ └────────────────────────────────────────┘ │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Tool Layer │ │ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │ │ │ API │ │ Database│ │ Functions │ │ │ │ Calls │ │ Query │ │ Execution │ │ │ └─────────┘ └─────────┘ └─────────────┘ │ └─────────────────────────────────────────────┘ 2. Planner模块 规划器负责将用户目标分解为可执行的步骤。 ...

引言：Go并发的优势与挑战 Go 语言的并发模型（CSP：Communicating Sequential Processes）通过 goroutine 和 channel 实现了"用同步的方式写异步代码"，极大降低了并发编程的门槛。 但在实际的高并发场景中，我们仍会面临： goroutine 泄漏：goroutine 数量失控 channel 死锁：协程间通信阻塞 GC 压力：内存分配频繁 CPU 占用高：上下文切换频繁 本文将从实战角度，系统介绍 Go 高并发服务的性能优化方法。 一、并发模型与调度器 1.1 Go 调度器原理 GMP 模型： ┌──────────────────────────────────────────────┐ │ P (Processor) │ │ (逻辑处理器，绑定 OS 线程) │ │ ┌──────────────┬──────────────┬────────────┐ │ │ │ G (goroutine) │ │ │ │ │ │ │ │ │ │ │ │ │ G │ G │ │ │ └──────────────┴──────────────┴────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────┐ │ │ │ M (Machine/OS Thread) │ │ │ └────────────────────────────────────────┘ │ └──────────────────────────────────────────────┘ GMP 参数调优： ...

引言：微服务治理的挑战 随着微服务数量的增长，服务治理成为不可回避的问题。当一个系统从 10 个服务增长到 100 个、甚至 1000 个服务时，以下问题会日益凸显： 服务发现：如何动态感知服务的上下线？ 负载均衡：如何将流量均匀分配到健康实例？ 故障隔离：如何防止级联故障（雪崩效应）？ 流量控制：如何保护系统不被突发流量打垮？ 灰度发布：如何安全地发布新版本？ 本文将系统地介绍服务治理的理论与实践。 一、服务注册与发现 1.1 服务注册中心选型 特性 Eureka Consul Nacos Etcd CAP AP CP AP/CP CP 一致性协议 最终一致 Raft Raft/Distro Raft 健康检查 客户端心跳 TCP/HTTP/gRPC TCP/HTTP/gRPC Lease 负载均衡 Ribbon 内置 内置 需集成 适用场景 通用 强一致性 通用 Kubernetes 1.2 Nacos 服务注册实战 服务端配置： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 # application.yml (Nacos Server) spring: datasource: platform: mysql mode: mysql num: 1 user: root password: password url: jdbc:mysql://127.0.0.1:3306/nacos_config?characterEncoding=utf8 nacos: raft: metadata: port: 8848 客户端注册： ...

引言：一致性难题的本质 在单体应用中，数据库事务（ACID）为我们提供了强大的一致性保证。但在分布式系统中，网络分区、节点故障、时钟漂移等不确定性因素，使得强一致性成为奢侈的选择。 如何在不同业务场景下选择合适的一致性方案？本文将从理论出发，结合实际代码，系统地介绍分布式一致性的工程实践。 一、理论基石：理解 CAP 与 BASE 1.1 CAP 定理的实践解读 一致性 (Consistency) ↗ ↖ / \ / \ 可用性 分区容错性 (Availability) (Partition Tolerance) 核心认知：在分布式系统中，P（分区容错）是客观存在的。我们真正选择的是在发生分区时，是选择 C（一致性）还是 A（可用性）。 场景分类： 业务场景 一性性要求 可用性要求 典型方案 金融转账 强一致性 可降级 2PC/TCC + 冲突等待 订单支付 最终一致性 高可用 Saga + 本地消息表 社交点赞 最终一致性 高可用 异步复制 + 冲突处理 库存扣减 强一致性 可降级 分布式锁 + Redis 1.2 BASE 理论的工程实践 BASE（Basically Available, Soft state, Eventually consistent）是对 CAP 中 AP 场景的补充： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 // 软状态示例：订单创建后，进入"处理中"状态 type Order struct { ID string Status OrderStatus // 软状态：pending -> confirmed -> shipped CreatedAt time.Time UpdatedAt time.Time // 状态会持续变化 } type OrderStatus int const ( StatusPending OrderStatus = iota // 初始状态 StatusConfirmed // 中间状态（软状态） StatusShipped // 最终状态 ) // 最终一致性：经过一段时间后，所有副本的状态会收敛 func (s *OrderService) ConfirmOrder(id string) error { // 1. 更新订单状态（立即返回） if err := s.repo.UpdateStatus(id, StatusConfirmed); err != nil { return err } // 2. 异步触发后续流程（最终一致性） go func() { s.inventoryClient.Deduct(id) // 可能失败，重试 s.shippingClient Arrange(id) // 可能失败，重试 s.notificationClient.Notify(id) // 可能失败，重试 }() return nil } 二、强一致性方案：2PC 与 3PC 2.1 两阶段提交（2PC）详解 架构： ...

引言：为什么需要缓存？ 在一个典型的电商系统中，读操作占比往往超过 80%。如果每次请求都打到数据库，单表的 QPS 上限大约在 1000-5000（取决于硬件配置）。当并发量达到 10 万+ 时，数据库必然成为瓶颈。 缓存是解决读性能问题的银弹，但它也带来了一系列挑战：数据一致性、缓存雪崩、缓存击穿、缓存穿透。本文将深入探讨这些问题及其解决方案。 一、缓存架构的演进路径 阶段1：无缓存时代 ┌─────────┐ │ 客户端 │ └────┬────┘ │ ▼ ┌─────────┐ │Web服务 │ └────┬────┘ │ ▼ ┌─────────┐ │ 数据库 │ ← 瓶颈 └─────────┘ 问题：数据库成为唯一瓶颈，QPS 上限低。 阶段2：单一 Redis 缓存 ┌─────────┐ │ 客户端 │ └────┬────┘ │ ▼ ┌─────────┐ │Web服务 │ └────┬────┘ │ ├───────────┐ ▼ ▼ ┌─────────┐ ┌─────────┐ │ Redis │ │ 数据库 │ └─────────┘ └─────────┘ 实现： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 type CacheService struct { redis *redis.Client db *sql.DB } func (s *CacheService) GetUser(ctx context.Context, userID string) (*User, error) { // 1. 尝试从 Redis 获取 cached, err := s.redis.Get(ctx, fmt.Sprintf("user:%s", userID)).Result() if err == nil { var user User json.Unmarshal([]byte(cached), &user) return &user, nil } // 2. Redis 未命中，查询数据库 user := &User{} if err := s.db.QueryRow("SELECT * FROM users WHERE id = ?", userID). Scan(user); err != nil { return nil, err } // 3. 写入 Redis data, _ := json.Marshal(user) s.redis.Set(ctx, fmt.Sprintf("user:%s", userID), data, 30*time.Minute) return user, nil } 问题： ...

引言：从监控到可观测性 传统的监控系统（如 Nagios、Zabbix）专注于基础设施监控：CPU、内存、磁盘、网络。但在容器化和微服务架构下，这些远远不够。 可观测性（Observability）的三大支柱： Metrics（指标）：数值化的时间序列数据，告警基础 Tracing（追踪）：请求在分布式系统中的完整路径 Logging（日志）：离散事件的记录，问题诊断 本文将介绍如何在 Kubernetes 环境下构建完整的可观测性体系。 一、Metrics：从黑盒到白盒 1.1 四大黄金指标 Latency（延迟）：请求的响应时间 Traffic（流量）：每秒请求数（QPS） Errors（错误率）：失败请求的百分比 Saturation（饱和度）：资源使用率 1.2 Prometheus + Grafana 监控架构 ┌────────────────────────────────────────────────────────┐ │ Grafana │ │ （统一可视化大盘） │ └────────────────────────────────────────────────────────┘ ↑ │ ┌────────────────────────────────────────────────────────┐ │ Prometheus │ │ （指标采集 + 存储 + 告警） │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Service A │ │ Service B │ │ Service C │ │ │ │ /metrics │ │ /metrics │ │ /metrics │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ └─────────┼────────────────┼────────────────┼───────────┘ │ │ │ └────────────────┴────────────────┘ ↓ ┌──────────┐ │ AlertManager │ │ （告警路由） │ └──────────┘ 1.3 自定义 Metrics：Go + Prometheus 依赖： ...

引言：为什么需要分库分表？ MySQL 单表性能瓶颈： 数据量：单表超过 1000 万行，查询性能显著下降 索引大小：索引树高度增加，磁盘 I/O 增多 锁竞争：高并发下锁等待严重 备份恢复：单表过大，备份耗时长 分库分表是突破单机数据库瓶颈的有效手段，但它也带来了复杂的路由逻辑和运维成本。本文将系统地介绍分库分表的实践方案。 一、垂直拆分 vs 水平拆分 1.1 垂直拆分（按业务维度） 单体数据库： ┌─────────────────────────────────────────┐ │ single_database │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ user │ │ order │ │ product │ │ │ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────┘ ↓ 垂直拆分 ┌─────────┐ ┌─────────┐ ┌─────────┐ │user_db │ │order_db │ │product_db│ │ │ │ │ │ │ │ user │ │ order │ │ product │ │ profile │ │ payment │ │ inventory│ └─────────┘ └─────────┘ └─────────┘ 优点： ...