Zhang XiJun

graph TD
    A[并发性] --> B[多个任务同时执行]
    C[共享性] --> D[资源被多个任务共同使用]
    E[异步性] --> F[任务执行“走走停停”]

    B & D & F --> G[现代操作系统的核心特征]
    G --> H[实现多任务、多用户环境]

gantt
    title 单道程序 vs 多道程序
    dateFormat  HH:mm:ss
    axisFormat  %Ss

    section 单道程序
    作业A-CPU     :crit, a1, 00:00:00, 2s
    作业A-I/O     :active, a2, after a1, 8s
    作业A-CPU     :crit, a3, after a2, 2s
    作业A-I/O     :active, a4, after a3, 8s

    section 多道程序
    作业A-CPU     :crit, b1, 00:00:00, 2s
    作业B-CPU     :crit, b2, after b1, 2s
    作业C-CPU     :crit, b3, after b2, 2s
    作业A-I/O     :active, b4, after b1, 8s
    作业B-I/O     :active, b5, after b2, 8s
    作业C-I/O     :active, b6, after b3, 8s

graph LR
    A[高级调度] -->|选作业进内存| B[中级调度]
    B -->|选进程进出内存| C[低级调度]
    C -->|选进程分配 CPU| D[CPU 执行]

    style A fill:#f9d5e5,stroke:#333
    style B fill:#e3eaa7,stroke:#333
    style C fill:#b2d3c2,stroke:#333

int turn;           // turn 表示轮到谁进入
boolean flag[2];   // flag[i] 表示进程 i 想进入临界区

// 初始化
flag[0] = flag[1] = false;

Process P0() {
    flag[0] = true;
    turn = 1;       // 谦让给 P1
    while (flag[1] && turn == 1); // 等待 P1 退出或谦让
    /* critical section */
    flag[0] = false;
    /* remainder section */
}

Process P1() {
    flag[1] = true;
    turn = 0;       // 谦让给 P0
    while (flag[0] && turn == 0); // 等待 P0 退出或谦让
    /* critical section */
    flag[1] = false;
    /* remainder section */
}

// P 操作 (Wait)
void P(semaphore s) {
    s.value = s.value - 1;
    if (s.value < 0) {
        // 资源不足，将当前进程加入等待队列并阻塞
        block(current_process);
    }
}

// V 操作 (Signal)
void V(semaphore s) {
    s.value = s.value + 1;
    if (s.value <= 0) {
        // 有进程在等待，唤醒一个
        wakeup(one_waiting_process);
    }
}

import threading
import time

# 5 把叉子（信号量）
forks = [threading.Semaphore(1) for _ in range(5)]
# 限制同时就餐人数为 4
dining_room = threading.Semaphore(4)

def philosopher(i):
    while True:
        think()
        dining_room.acquire()  # 进入餐厅（最多 4 人）
        
        forks[i].acquire()       # 拿起左边叉子
        forks[(i + 1) % 5].acquire()  # 拿起右边叉子
        
        eat(i)
        
        forks[i].release()       # 放下左边叉子
        forks[(i + 1) % 5].release()  # 放下右边叉子
        
        dining_room.release()    # 离开餐厅

def think():
    time.sleep(0.1)

def eat(i):
    print(f"Philosopher {i} is eating...")
    time.sleep(0.1)

# 创建 5 个哲学家线程
threads = []
for i in range(5):
    t = threading.Thread(target=philosopher, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

def philosopher(i):
    while True:
        think()
        
        if i % 2 == 0:  # 偶数号哲学家
            forks[(i + 1) % 5].acquire()  # 先拿右边叉子
            forks[i].acquire()             # 再拿左边叉子
        else:  # 奇数号哲学家
            forks[i].acquire()             # 先拿左边叉子
            forks[(i + 1) % 5].acquire()  # 再拿右边叉子
        
        eat(i)
        
        forks[i].release()                 # 放下左边叉子
        forks[(i + 1) % 5].release()      # 放下右边叉子

item B[n];
Semaphore empty;  /*可用的空缓冲区个数*/
Semaphore full;   /*可用的产品数*/
Semaphore mutex;  /*互斥信号量*/
empty = n; full = 0; mutex = 1;
int in = 0; out = 0;  /*in为放入缓冲区指针, out为取出缓冲区指针*/

Process producer_i( ) {
    while(true) {
        item product = produce();
        P(empty);
        P(mutex);
        B[in] = product;
        in = (in+1) % n;
        V(mutex);
        V(full);
    }
}

Process consumer_i( ) {
    while(true) {
        P(full);
        P(mutex);
        Item product = B[out];
        out = (out+1) % n;
        V(mutex);
        V(empty);
        consume(product);
    }
}

sequenceDiagram
    participant Producer as 生产者 P1
    participant Consumer as 消费者 C1
    participant Mutex as 互斥锁 (mutex)
    participant Empty as 空闲缓冲区 (empty)

    Note over Producer,Consumer: 初始状态: empty=0 (缓冲区满), mutex=1

    Producer->>Mutex: P(mutex) // 成功获取锁，mutex=0
    Producer->>Empty: P(empty) // empty=0，阻塞！等待空位...
    Note over Producer: 生产者 P1 霸占 mutex 锁，等待 empty

    Consumer->>Full: P(full) // full=1，成功，full=0
    Consumer->>Mutex: P(mutex) // mutex=0，阻塞！等待锁...
    Note over Consumer: 消费者 C1 等待 mutex 锁

    Note over Producer,Consumer: 💥 死锁！
    Note right of Producer: 我要等 empty (需 C1 消费)
    Note left of Consumer: 我要等 mutex (需 P1 释放)

Semaphore pmutex, cmutex; // 两个独立的互斥锁
...
P(pmutex); // 生产者只锁自己的写入区域
P(cmutex); // 消费者只锁自己的读取区域

# Download the configuration file and rename it as docker-compose.yml
C:\>Invoke-WebRequest https://github.com/milvus-io/milvus/releases/download/v2.6.0/milvus-standalone-docker-compose.yml -OutFile docker-compose.yml

# Start Milvus
C:\>docker compose up -d

pip install -U pymilvus

class MilvusStorage:
    """Milvus向量存储管理类
    
    负责将分块后的文档内容存储到Milvus向量数据库中，
    支持向量检索和BM25全文检索。
    """
    
    def __init__(self, 
                 embedding_function: Embeddings,
                 uri: Optional[str] = None, 
                 db_name: Optional[str] = None,
                 token: Optional[str] = None,
                 collection_name: Optional[str] = None):
        """初始化Milvus存储客户端
        
        Args:
            embedding_function: LangChain embedding模型实例（必需）
            uri: Milvus服务地址，默认从环境变量MILVUS_URI获取
            db_name: 数据库名称，默认从环境变量MILVUS_DB_NAME获取
            token: 认证令牌，默认从环境变量MILVUS_TOKEN获取（可选）
            collection_name: 集合名称，默认从环境变量MILVUS_COLLECTION_NAME获取
        """
        # 验证必需参数
        if not embedding_function:
            raise ValueError("embedding_function是必需参数，必须提供LangChain Embeddings实例")
        
        # 从环境变量读取配置，如果参数没有提供的话
        self.uri = uri or os.getenv('MILVUS_URI', 'http://localhost:19530')
        self.db_name = db_name or os.getenv('MILVUS_DB_NAME', 'rag')
        self.token = token or os.getenv('MILVUS_TOKEN') or None
        self.collection_name = collection_name or os.getenv('MILVUS_COLLECTION_NAME', 'chunks')
        
        # 设置embedding函数
        self.embedding_function = embedding_function
        
        # 初始化LangChain Milvus向量存储
        self.vector_store = Milvus(
            embedding_function=self.embedding_function,
            connection_args={
                "uri": self.uri,
                "db_name": self.db_name,
                "token": self.token
            } if self.token else {
                "uri": self.uri,
                "db_name": self.db_name
            },
            collection_name=self.collection_name,
            index_params={"index_type": "HNSW", "metric_type": "COSINE", "params": {"M": 16, "efConstruction": 200}}
        )
        
    def store_chunks(self, chunk_result: ChunkResult) -> Dict[str, Any]:
        """存储分块结果到Milvus
        
        Args:
            chunk_result: 分块结果对象
            
        Returns:
            Dict: 插入结果，包含插入状态和记录数
            
        Raises:
            ValueError: 当向量存储未初始化时
            Exception: Milvus操作异常
        """
        if not self.vector_store:
            raise ValueError("向量存储未初始化")
        
        if not chunk_result.chunks:
            return {
                "status": "success", 
                "inserted_count": 0,
                "message": "无数据需要插入"
            }
        
        try:
            # 转换为LangChain Document格式
            documents = self._convert_chunks_to_langchain_docs(chunk_result)
            
            # 为每个文档生成UUID作为主键
            from uuid import uuid4
            uuids = [str(uuid4()) for _ in range(len(documents))]
            
            # 使用LangChain Milvus添加文档，指定IDs
            ids = self.vector_store.add_documents(documents=documents, ids=uuids)
            
            return {
                "status": "success",
                "inserted_count": len(documents),
                "document_ids": ids,
                "document_name": chunk_result.document_name,
                "strategy": chunk_result.strategy.value,
                "collection_name": self.collection_name
            }
            
        except Exception as e:
            raise Exception(f"Milvus插入失败: {str(e)}")
    
    def _convert_chunks_to_langchain_docs(self, chunk_result: ChunkResult) -> List[Document]:
        """为现有Documents添加存储所需的元数据
        
        Args:
            chunk_result: 分块结果，chunks已经是Document列表
            
        Returns:
            List[Document]: 添加了元数据的Document列表
        """
        documents = []
        
        for idx, chunk in enumerate(chunk_result.chunks):
            # 创建符合Milvus集合schema的元数据
            # 注意：page_content会自动映射到text_content字段
            updated_metadata = {
                **chunk.metadata,  # 保留原有元数据
                "document_name": chunk_result.document_name,
                "chunk_index": idx,
                "chunk_size": len(chunk.page_content)
            }
            
            # 创建新Document以避免修改原始数据
            # LangChain会自动将page_content映射到Milvus的text_content字段
            # embedding字段会由embedding_function自动生成
            doc = Document(
                page_content=chunk.page_content,
                metadata=updated_metadata
            )
            documents.append(doc)
        
        return documents
    
    def store_chunks_batch(self, chunk_results: List[ChunkResult]) -> Dict[str, Any]:
        """批量存储多个分块结果到Milvus
        
        Args:
            chunk_results: 分块结果列表
            
        Returns:
            Dict: 批量插入结果
            
        Raises:
            ValueError: 当向量存储未初始化时
            Exception: Milvus操作异常
        """
        if not self.vector_store:
            raise ValueError("向量存储未初始化")
        
        if not chunk_results:
            return {
                "status": "success",
                "message": "没有分块结果需要存储",
                "total_chunks": 0,
                "document_count": 0
            }
        
        try:
            # 收集所有文档
            all_documents = []
            total_chunks = 0
            
            for chunk_result in chunk_results:
                if chunk_result.chunks:
                    documents = self._convert_chunks_to_langchain_docs(chunk_result)
                    all_documents.extend(documents)
                    total_chunks += len(documents)
            
            if not all_documents:
                return {
                    "status": "success",
                    "message": "没有文档需要存储",
                    "total_chunks": 0,
                    "document_count": len(chunk_results)
                }
            
            # 为所有文档生成UUID作为主键
            from uuid import uuid4
            uuids = [str(uuid4()) for _ in range(len(all_documents))]
            
            # 批量添加所有文档，指定IDs
            ids = self.vector_store.add_documents(documents=all_documents, ids=uuids)
            
            return {
                "status": "success",
                "message": f"成功存储 {len(chunk_results)} 个文档的 {total_chunks} 个分块",
                "total_chunks": total_chunks,
                "document_count": len(chunk_results),
                "ids": ids,
                "collection_name": self.collection_name
            }
            
        except Exception as e:
            raise Exception(f"Milvus批量插入失败: {str(e)}")
    
    def delete_document(self, 
                       document_name: str, 
                       collection_name: Optional[str] = None) -> Dict[str, Any]:
        """删除指定文档的所有chunks
        
        Args:
            document_name: 文档名称
            collection_name: collection名称
            
        Returns:
            Dict: 删除结果
        """
        target_collection = collection_name or self.collection_name
        
        try:
            if not self.vector_store:
                raise ValueError("向量存储未初始化")
            
            # 使用LangChain Milvus删除功能
            # 注意：LangChain Milvus可能不支持按元数据过滤删除，这里提供基本实现
            return {
                "status": "error",
                "document_name": document_name,
                "error": "LangChain Milvus不支持按文档名删除，请使用其他方式",
                "collection_name": target_collection
            }
            
        except Exception as e:
            return {
                "status": "error", 
                "document_name": document_name,
                "error": str(e)
            }
    
    def get_document_stats(self, 
                          document_name: Optional[str] = None,
                          collection_name: Optional[str] = None) -> Dict[str, Any]:
        """获取文档统计信息
        
        Args:
            document_name: 文档名称，None则统计所有文档
            collection_name: collection名称
            
        Returns:
            Dict: 统计信息
        """
        target_collection = collection_name or self.collection_name
        
        try:
            if not self.vector_store:
                raise ValueError("向量存储未初始化")
            
            # 使用LangChain Milvus获取基本信息
            # 注意：LangChain Milvus没有直接的统计方法，这里提供基本信息
            return {
                "status": "success",
                "collection_name": target_collection,
                "vector_store_type": "LangChain Milvus",
                "embedding_function": str(type(self.embedding_function).__name__),
                "connection_uri": self.uri,
                "database_name": self.db_name,
                "message": "详细统计信息需要通过其他方式获取"
            }
                
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

flowchart TB
    subgraph "Plan-and-Execute阶段"
        A["用户输入"] --> B["Planner Agent"]
        B --> C["Agent Executor"]
        C --> D["Replanner"]
        D -->|"需要更多信息"| C
        D -->|"信息充足"| E["输出结构化信息"]
    end
    
    subgraph "内容生成阶段"
        E --> F["大纲设计节点"]
        F --> G["页面内容生成节点"]
        G --> H["HTML代码生成节点"]
    end
    
    subgraph "文件转换阶段"
        H --> I["html演示生成"]
        I --> J["转换pptx节点"]
        J --> K["输出PPTX文件"]
    end

from apryse_sdk import PDFNet, PDFDoc, Convert, StructuredOutputModule

# 1. 初始化（许可证）
PDFNet.Initialize("demo:1756369085114:")

# 2. 告诉 SDK 模块放在哪里
PDFNet.AddResourceSearchPath(r"F:\project python\test\StructuredOutputWindows\Lib\Windows")

# 3. 可选：确认模块已就位
if not StructuredOutputModule.IsModuleAvailable():
    raise RuntimeError("StructuredOutput module not found!")

# 4. 正常调用
doc = PDFDoc("input.pdf")
Convert.ToPowerPoint(doc, "output.pptx")

from langchain_community.document_loaders import WebBaseLoader

urls = [
    "https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",
    "https://lilianweng.github.io/posts/2024-07-07-hallucination/",
    "https://lilianweng.github.io/posts/2024-04-12-diffusion-video/",
]

docs = [WebBaseLoader(url).load() for url in urls]

docs[0][0].page_content.strip()[:1000]

from langchain_text_splitters import RecursiveCharacterTextSplitter

docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

doc_splits[0].page_content.strip()

from langchain_chroma import Chroma  # 导入 Chroma
from langchain_openai import OpenAIEmbeddings
import os

# 确保安装了 langchain-chroma
# pip install langchain-chroma

embedding = OpenAIEmbeddings(
    api_key="sk-", 
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="text-embedding-v4",
    check_embedding_ctx_length=False,
    dimensions=1536,
    chunk_size=5  # 设置较小的批次大小
)

# 使用 Chroma 替代 InMemoryVectorStore
vectorstore = Chroma.from_documents(
    documents=doc_splits, 
    embedding=embedding,
    persist_directory="./chroma_db"  # 指定持久化目录
)

# 重新加载已存在的 Chroma 数据库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding
)

retriever = vectorstore.as_retriever()

from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever,                    # 【输入】一个已经配置好的检索器（例如：向量数据库的检索器）
    "retrieve_blog_posts",        # 【工具名称】这个工具的唯一标识名（供模型内部调用）
    "Search and return information about Lilian Weng blog posts."  # 【工具描述】模型看到的说明，用于决定是否调用它
)

retriever_tool.invoke({"query": "types of reward hacking"})

from langchain_community.chat_models import ChatTongyi
llm = ChatTongyi(
    model="qwen3-235b-a22b",
    api_key="sk-",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model_kwargs={"enable_thinking": False}   # 关键在这里
)

from langgraph.graph import MessagesState

def generate_query_or_respond(state: MessagesState):
    """调用模型，根据当前状态生成响应。根据问题，模型将决定是使用检索工具进行检索，还是直接回复用户。
    """
    response = (
        llm
        .bind_tools([retriever_tool]).invoke(state["messages"])
    )
    return {"messages": [response]}

input = {
    "messages": [
        {
            "role": "user",
            "content": "What does Lilian Weng say about types of reward hacking?",
        }
    ]
}
generate_query_or_respond(input)["messages"][-1].pretty_print()

from pydantic import BaseModel, Field
from typing import Literal

# 定义评分提示模板
GRADE_PROMPT = (
    "你是一个评分员，负责评估检索到的文档与用户问题的相关性。\n "
    "以下是检索到的文档内容：\n\n {context} \n\n"
    "以下是用户的问题：{question} \n"
    "如果文档包含与用户问题相关的关键词或语义含义，则将其评为相关。\n"
    "请给出一个二元评分：'yes'（是）表示相关，'no'（否）表示不相关。"
)

# 定义用于评估文档相关性的 Pydantic 模型
class GradeDocuments(BaseModel):
    """使用二元评分对文档进行相关性评估。"""
    
    binary_score: str = Field(
        description="相关性评分：'yes' 表示相关，'no' 表示不相关"
    )

# 初始化用于评分的聊天模型
grader_model = llm

def grade_documents(
    state: MessagesState,
) -> Literal["generate_answer", "rewrite_question"]:
    """
    判断检索到的文档是否与用户问题相关。
    
    参数:
        state: 包含消息历史的状态对象，其中第一条消息是用户问题，
               最后一条消息是检索到的文档内容。
    
    返回:
        如果文档相关，返回 "generate_answer"；
        如果不相关，返回 "rewrite_question"，表示需要重写问题并重新检索。
    """
    question = state["messages"][0].content  # 获取用户问题
    context = state["messages"][-1].content  # 获取检索到的文档内容

    # 将问题和文档内容填入提示模板
    prompt = GRADE_PROMPT.format(question=question, context=context)
    
    # 调用模型，并以结构化输出（Pydantic 模型）的形式获取评分结果
    response = (
        grader_model
        .with_structured_output(GradeDocuments)
        .invoke([{"role": "user", "content": prompt}])
    )
    #print(response)
    score = response.binary_score  # 获取二元评分结果
 
    # 根据评分决定下一步操作
    if score == "yes":
        return "generate_answer"  # 文档相关，生成答案
    else:
        return "rewrite_question"  # 文档不相关，重写问题后重新检索

from langchain_core.messages import convert_to_messages

input = {
    "messages": convert_to_messages(#将一系列消息转换为 BaseMessage 类型的消息列表。
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {"role": "tool", "content": "meow", "tool_call_id": "1"},
        ]
    )
}
grade_documents(input)

REWRITE_PROMPT = (
    "Look at the input and try to reason about the underlying semantic intent / meaning.\n"
    "Here is the initial question:"
    "\n ------- \n"
    "{question}"
    "\n ------- \n"
    "Formulate an improved question:"
)

def rewrite_question(state: MessagesState):
    """
    重写用户最初的提问，以更好地表达其语义意图。
    
    参数:
        state: 包含消息历史的状态对象，其中第一条消息是用户原始问题。
    
    返回:
        一个字典，包含一条新的用户消息，内容为改写后的问题。
        该消息将用于后续的检索步骤，以提高检索结果的相关性。
    """
    messages = state["messages"]
    question = messages[0].content  # 获取用户最初的提问
    prompt = REWRITE_PROMPT.format(question=question)  # 将问题填入提示模板
    response = llm.invoke([{"role": "user", "content": prompt}])  # 调用模型生成改写后的问题
    
    # 返回新的消息结构，内容为改写后的问题
    return {"messages": [{"role": "user", "content": response.content}]}

input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {"role": "tool", "content": "meow", "tool_call_id": "1"},
        ]
    )
}

response = rewrite_question(input)
print(response["messages"][-1]["content"])

GENERATE_PROMPT = (
    "You are an assistant for question-answering tasks. "
    "Use the following pieces of retrieved context to answer the question. "
    "If you don't know the answer, just say that you don't know. "
    "Use three sentences maximum and keep the answer concise.\n"
    "Question: {question} \n"
    "Context: {context}"
)


def generate_answer(state: MessagesState):
    """Generate an answer."""
    question = state["messages"][0].content
    context = state["messages"][-1].content
    prompt = GENERATE_PROMPT.format(question=question, context=context)
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response]}

input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {
                "role": "tool",
                "content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
                "tool_call_id": "1",
            },
        ]
    )
}

response = generate_answer(input)
response["messages"][-1].pretty_print()

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition

# 创建一个基于状态图（StateGraph）的流程，用于管理对话或任务的执行流程
workflow = StateGraph(MessagesState)

# 定义流程中将循环执行的各个节点
workflow.add_node(generate_query_or_respond)          # 判断是生成检索查询还是直接回复用户
workflow.add_node("retrieve", ToolNode([retriever_tool]))  # 检索节点：使用检索工具（retriever_tool）从知识库中查找相关文档
workflow.add_node(rewrite_question)                  # 重写问题节点：当检索结果不相关时，优化并重写用户的问题
workflow.add_node(generate_answer)                   # 生成答案节点：基于检索到的信息生成最终回答

# 设置流程的起始点：从 `generate_query_or_respond` 节点开始
workflow.add_edge(START, "generate_query_or_respond")

# 添加条件边：决定是否进行文档检索
workflow.add_conditional_edges(
    "generate_query_or_respond",
    # 使用 `tools_condition` 函数判断 LLM 的输出意图：
    # 如果 LLM 决定调用 `retriever_tool` 工具，则进入检索；如果选择直接回复，则结束流程
    tools_condition,
    {
        # 将条件判断结果映射到图中的具体节点
        "tools": "retrieve",   # 若需调用工具，则跳转到检索节点
        END: END               # 若无需调用工具（即可以直接回答），则结束流程
    },
)

# 在 `retrieve` 节点执行后，根据文档相关性判断下一步操作
workflow.add_conditional_edges(
    "retrieve",
    # 调用 `grade_documents` 函数评估检索到的文档是否与问题相关
    grade_documents,
    # 根据评分结果决定流向：
    # - 如果相关，进入 `generate_answer`
    # - 如果不相关，进入 `rewrite_question`
    # （该逻辑在 `grade_documents` 函数中返回 "generate_answer" 或 "rewrite_question"）
)

# 添加固定边：生成答案后流程结束
workflow.add_edge("generate_answer", END)

# 重写问题后，回到初始节点重新判断是否需要检索
workflow.add_edge("rewrite_question", "generate_query_or_respond")

# 编译整个工作流，生成可执行的图结构
graph = workflow.compile()

pip install -U langgraph langgraph-supervisor langchain-tavily "langchain[openai]"

from langchain_tavily import TavilySearch

web_search = TavilySearch(max_results=3,tavily_api_key="tvly-dev-")
web_search_results = web_search.invoke("南京在哪")

print(web_search_results["results"][0]["content"])

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm=ChatOpenAI(
    model="qwen3-235b-a22b-thinking-2507",
    api_key="sk-",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

research_agent = create_react_agent(
    model=llm,
    tools=[web_search],
    prompt=(
        "你是一个研究代理。\n\n指令：\n- 仅协助与研究相关的任务，不得进行任何数学计算\n- 完成任务后，直接向主管回复\n- 仅回复你的工作结果，不得包含任何其他文字。"
    ),
    name="research_agent",
)

from langchain_core.messages import convert_to_messages


def pretty_print_message(message, indent=False):
    """
    美化打印单条消息
    
    Args:
        message: 要打印的消息对象
        indent: 是否需要缩进打印
    """
    # 将消息转换为美观的HTML格式表示
    pretty_message = message.pretty_repr(html=True)
    if not indent:
        # 如果不需要缩进，直接打印
        print(pretty_message)
        return

    # 如果需要缩进，为每一行添加制表符前缀
    indented = "\n".join("\t" + c for c in pretty_message.split("\n"))
    print(indented)


def pretty_print_messages(update, last_message=False):
    """
    美化打印消息更新
    
    Args:
        update: 包含消息更新的数据结构
        last_message: 是否只打印最后一条消息
    """
    is_subgraph = False  # 标记是否为子图更新
    
    # 检查更新是否为元组格式（包含命名空间信息）
    if isinstance(update, tuple):
        ns, update = update
        # 如果命名空间为空，跳过父图更新的打印
        if len(ns) == 0:
            return

        # 提取图ID并打印子图更新信息
        graph_id = ns[-1].split(":")[0]
        print(f"来自子图 {graph_id} 的更新:")
        print("\n")
        is_subgraph = True

    # 遍历每个节点的更新
    for node_name, node_update in update.items():
        # 构造更新标签
        update_label = f"来自节点 {node_name} 的更新:"
        if is_subgraph:
            # 如果是子图，添加缩进
            update_label = "\t" + update_label

        print(update_label)
        print("\n")

        # 将节点更新中的消息转换为消息对象列表
        messages = convert_to_messages(node_update["messages"])
        # 如果只要求最后一条消息，则截取最后一条
        if last_message:
            messages = messages[-1:]

        # 打印每条消息
        for m in messages:
            pretty_print_message(m, indent=is_subgraph)
        print("\n")

for chunk in research_agent.stream(
    {"messages": [{"role": "user", "content": "南京在哪?"}]}
):
    pretty_print_messages(chunk)

def add(a: float, b: float):
    """将两个数字相加。"""
    return a + b


def multiply(a: float, b: float):
    """将两个数字相乘。"""
    return a * b


def divide(a: float, b: float):
    """将两个数字相除。"""
    return a / b


math_agent = create_react_agent(
    model=llm,
    tools=[add, multiply, divide],
    prompt=(
        "你是一个数学代理。\n\n"
        "指令：\n"
        "- 仅协助处理数学相关任务\n"
        "- 完成任务后，直接回复给主管\n"
        "- 仅回复你的工作结果，不要包含任何其他文字。"
    ),
    name="math_agent",
)

for chunk in math_agent.stream(
    {"messages": [{"role": "user", "content": "what's (3 + 5) x 7"}]}
):
    pretty_print_messages(chunk)

from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, math_agent],
    prompt=(
        "你是一个管理两个代理的主管：\n"
        "- 一个研究代理。将研究相关任务分配给这个代理\n"
        "- 一个数学代理。将数学相关任务分配给这个代理\n"
        "一次只分配工作给一个代理，不要并行调用代理。\n"
        "不要自己做任何工作。"
    ),
    add_handoff_back_messages=True,
    output_mode="full_history",
).compile()

from IPython.display import display, Image

display(Image(supervisor.get_graph().draw_mermaid_png()))

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command


def create_handoff_tool(*, agent_name: str, description: str | None = None):
    """
    创建一个“交接”工具函数，用于在 LangGraph 的 Supervisor-Worker 架构中
    把当前对话状态移交给指定名称的子 Agent。

    参数
    ----
    agent_name : str
        目标子 Agent 的名称，必须与 Supervisor 图中注册的节点名一致。
    description : str | None
        工具的描述文本。如果为 None，则使用默认描述 "Ask {agent_name} for help."。

    返回
    ----
    handoff_tool : Callable
        一个已用 @tool 装饰的函数，可直接注入到 Supervisor 的工具列表。
    """
    # 动态生成工具名，例如 agent_name="math_agent" -> "transfer_to_math_agent"
    name = f"transfer_to_{agent_name}"

    # 如果调用者没有提供描述，则使用默认描述
    description = description or f"Ask {agent_name} for help."

    # 用 LangGraph 的 @tool 装饰器注册工具
    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        """
        实际执行交接逻辑的工具函数。

        参数
        ----
        state : MessagesState
            当前对话状态，由 LangGraph 注入。
        tool_call_id : str
            本次工具调用的唯一 ID，由 LangGraph 注入。

        返回
        ----
        Command
            一个 LangGraph Command 对象，告诉框架：
            - goto=agent_name        : 跳转到哪个子 Agent
            - update                 : 更新后的状态
            - graph=Command.PARENT   : 在父图（Supervisor）作用域内执行
        """
        # 构造一条工具消息，记录交接动作
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }

        # 使用 Command 把对话状态连同新消息一起发送到目标 Agent
        return Command(
            goto=agent_name,
            update={**state, "messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )

    # 返回已装饰的工具函数，供 Supervisor 添加进 tools 列表
    return handoff_tool


# 创建研究代理的交接工具
assign_to_research_agent = create_handoff_tool(
    agent_name="research_agent",
    description="Assign task to a researcher agent.",
)

# 创建数学代理的交接工具
assign_to_math_agent = create_handoff_tool(
    agent_name="math_agent",
    description="Assign task to a math agent.",
)

supervisor_agent = create_react_agent(
    model=llm,
    tools=[assign_to_research_agent, assign_to_math_agent],
    prompt=(
        "你是一个管理两个代理的主管：\n"
        "- 一个研究代理。将研究相关任务分配给这个代理\n"
        "- 一个数学代理。将数学相关任务分配给这个代理\n"
        "一次只分配工作给一个代理，不要并行调用代理。\n"
        "不要自己做任何工作。"
    ),
    name="supervisor",
)

from langgraph.graph import END

# 定义多代理主管图
supervisor = (
    StateGraph(MessagesState)
    # 注意：`destinations` 仅用于可视化，不影响运行时行为
    .add_node(supervisor_agent, destinations=("research_agent", "math_agent", END))
    .add_node(research_agent)
    .add_node(math_agent)
    .add_edge(START, "supervisor")
    # 总是返回到主管
    .add_edge("research_agent", "supervisor")
    .add_edge("math_agent", "supervisor")
    .compile()
)

# 1. 定义研究团队的代理
@tool
def web_search(query: str) -> str:
    """执行网络搜索"""
    return f"搜索结果：关于'{query}'的最新信息..."

@tool
def analyze_data(data: str) -> str:
    """分析数据"""
    return f"数据分析结果：{data}的趋势显示..."

research_agent = create_react_agent(
    model=llm,
    tools=[web_search, analyze_data],
    prompt="你是一个研究专家，负责进行网络搜索和数据分析。",
    name="research_specialist"
)

# 2. 定义数学团队的代理
@tool
def calculate_statistics(numbers: list[float]) -> str:
    """计算统计值"""
    if not numbers:
        return "错误：数据列表为空"
    avg = sum(numbers) / len(numbers)
    return f"统计结果：平均值={avg:.2f}，数据点数量={len(numbers)}"

@tool
def solve_equation(equation: str) -> str:
    """解方程"""
    return f"方程 {equation} 的解为：x = 42"

math_agent = create_react_agent(
    model=llm,
    tools=[calculate_statistics, solve_equation],
    prompt="你是一个数学专家，负责统计计算和方程求解。",
    name="math_specialist"
)

# 3. 创建研究团队主管
research_supervisor = create_supervisor(
    model=llm,
    agents=[research_agent],
    prompt=(
        "你是研究团队的主管。\n"
        "你的团队有一个研究专家，负责网络搜索和数据分析。\n"
        "根据任务需求，将工作分配给研究专家。\n"
        "等待专家完成任务后，总结结果并报告给上级主管。"
    ),
    name="research_supervisor"
).compile(name="research_supervisor")

# 4. 创建数学团队主管
math_supervisor = create_supervisor(
    model=llm,
    agents=[math_agent],
    prompt=(
        "你是数学团队的主管。\n"
        "你的团队有一个数学专家，负责统计计算和方程求解。\n"
        "根据任务需求，将工作分配给数学专家。\n"
        "等待专家完成任务后，总结结果并报告给上级主管。"
    ),
    name="math_supervisor"
).compile(name="math_supervisor")

# 5. 创建顶层主管
top_supervisor = create_supervisor(
    model=llm,
    agents=[research_supervisor, math_supervisor],
    prompt=(
        "你是顶层主管，管理两个专业团队：\n"
        "- 研究团队：负责市场调研、数据分析等任务\n"
        "- 数学团队：负责统计计算、方程求解等任务\n"
        "根据任务的性质，将工作分配给相应的团队主管。\n"
        "等待团队完成任务后，整合所有结果并给出最终报告。"
    ),
    name="top_supervisor"
).compile(name="top_supervisor")

npx create-react-app my-app

my-app
├─ public/         # 静态资源，index.html 是页面模板
├─ src/
│  ├─ App.js       # 根组件
│  ├─ index.js     # 应用入口（ReactDOM.createRoot）
└─ package.json    # 依赖与脚本

// 1. 找到 public/index.html 中 id="root" 的 DOM 节点，作为 React 应用的挂载点
const root = ReactDOM.createRoot(document.getElementById('root'));

// 2. 将根组件 <App /> 渲染到该挂载点
root.render(
  // 3. <React.StrictMode> 是 React 提供的开发模式辅助工具
  //    作用：在开发阶段自动检测潜在问题（如过时的 API、副作用重复执行等）
  //    注意：它仅在开发环境生效，生产环境不会渲染任何额外 DOM
  <React.StrictMode>
    {/* 4. 项目真正的根组件 App，所有业务逻辑都从这里开始 */}
    <App />
  </React.StrictMode>
);

const name = 'React';
<h1>Hello, {name}!</h1>          // → Hello, React!

function App() {
  const mytitle="hello"
  return (
    <div title={mytitle}></div>
  );
}

function App() {
  const mytitle="hello"

  let mycontent=null
  const flag=true
  if(flag){
    mycontent=<h2>hello</h2>
  }
  else{
    mycontent=<h2>world</h2>
  }
  return (
    <div title={mytitle}>{mycontent}</div>
  );
}

function App() {
  const list=['1','2','3']
  const mycontent=list.map((item)=>{
    return <li>{item}</li>
  })
  return (
    <div>{mycontent}</div>
  );
}

import { useState } from 'react';
function App() {
  const [mycontent,setmycontent]=useState("hello world");
  function changeContent(){
    setmycontent("hello world2");
  }
  return (
    <>
      <div>{mycontent}</div>
      <button onClick={changeContent}>change</button>
    </>
  );
}