Zhang XiJun

# 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>
    </>
  );
}

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

import { useState } from 'react';
function App() {
return (
    <>
      <img src={logo} className="App-logo" alt="logo" style={{ width: '100px',backgroundColor: 'grey'}}/>
    </>
  );
}

function App() {

  const imgdata={
    className:"App-logo",
    style:{
      width:'100px',
      backgroundColor:'grey'
    }
  }

  return (
    <>
      <img src={logo} alt="logo" {...imgdata}/>
    </>
  );
}

function Article(props) {
  return (
    <div>
      <h2>{props.title}</h2>
      <p>{props.content}</p>
    </div>
  );
}

function App() {
  return (
    <>
      <Article title="标签1" content="内容1" />
      <Article title="标签2" content="内容2" />
    </>
  );
}

function Parent() {
  const title = 'Hello React';
  return <Child title={title} />;
}

function Child({ title }) {
  return <h1>{title}</h1>;
}

function Parent() {
  const [count, setCount] = useState(0);
  return (
    <>
      <p>父：{count}</p>
      <Child onInc={() => setCount(c => c + 1)} />
    </>
  );
}

function Child({ onInc }) {
  return <button onClick={onInc}>子按钮 +1</button>;
}

from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langgraph.utils.config import get_store 
from langmem import (
    # 让智能体创建、更新和删除记忆 
    create_manage_memory_tool,
)

def prompt(state):
    """为LLM准备消息。"""
    # 从配置的上下文变量中获取存储; 
    store = get_store() # 与提供给 `create_react_agent` 的相同
    memories = store.search(
        # 在与我们为智能体配置的相同命名空间内搜索
        ("memories",),
        query=state["messages"][-1].content,
    )
    system_msg = f"""You are a helpful assistant.

## Memories
<memories>
{memories}
</memories>
"""
    return [{"role": "system", "content": system_msg}, *state["messages"]]

from langchain import embeddings
from langchain_openai import OpenAIEmbeddings
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 
)
store = InMemoryStore(
    index={ # 存储提取的记忆 
        "dims": 1536,
        "embed": embedding,
    }
) 
checkpointer = MemorySaver() # 检查点图状态 

from langchain_openai import ChatOpenAI
model_qwen=ChatOpenAI(
    api_key="sk-", 
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen3-30b-a3b-instruct-2507",
)

agent = create_react_agent( 
    model=model_qwen,
    prompt=prompt,
    tools=[ # 添加记忆工具 
        # 智能体可以调用 "manage_memory" 来
        # 通过ID创建、更新和删除记忆
        # 命名空间为记忆添加作用域。要
        # 为每个用户限定记忆范围，使用 ("memories", "{user_id}"): 
        create_manage_memory_tool(namespace=("memories",)),
    ],
    # 我们的记忆将存储在这个提供的BaseStore实例中
    store=store,
    # 图的"状态"将在每个节点完成执行后进行检查点
    # 用于跟踪聊天历史和持久执行
    checkpointer=checkpointer, 
)

agent.get_graph().draw_mermaid_png(output_file_path="agent.png")

config = {"configurable": {"thread_id": "thread-a"}} 
agent.invoke( 
    { 
        "messages": [ 
            {"role": "user", "content": "我喜欢黑色的显示模式"} 
        ] 
    }, 
    # 我们将通过使用具有相同thread_id的config
    # 来继续对话(thread-a)
    config=config, 
) 
print(response["messages"][-1].content) 

是的，我知道！你偏好黑色显示模式。我会在后续交互中保持这一设置。

# 新线程 = 新对话！
new_config = {"configurable": {"thread_id": "thread-b"}} 
# 智能体只能回忆起
# 它使用manage_memories工具明确保存的内容
response = agent.invoke( 
    {"messages": [{"role": "user", "content": "你好。你还记得我吗？你知道我有什么偏好吗？"}]},
    config=new_config, 
) 
print(response["messages"][-1].content) 

你好！虽然我无法记住你作为个体的详细信息，但我可以访问一些关于你的偏好信息。根据之前的记录，我知道你偏好使用黑色显示模式。如果你还有其他偏好或希望我记住什么，请告诉我，我会帮你记录下来。

from langchain.chat_models import init_chat_model 
from langgraph.func import entrypoint 
from langgraph.store.memory import InMemoryStore 

from langmem import ReflectionExecutor, create_memory_store_manager 

from langchain_openai import OpenAIEmbeddings
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 
)
store = InMemoryStore(
    index={ # 存储提取的记忆 
        "dims": 1536,
        "embed": embedding,
    }
) 

# 创建记忆管理器 Runnable 来从对话中提取记忆
memory_manager = create_memory_store_manager( 
    model_qwen, 
    # 将记忆存储在 "memories" 命名空间（即目录）中
    namespace=("memories",),  
    instructions="用中文存储记忆。"
) 

# 包装 memory_manager 以处理延迟的后台处理
executor = ReflectionExecutor(memory_manager) 

from langchain_openai import ChatOpenAI
model_qwen=ChatOpenAI(
    api_key="sk-", 
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen3-30b-a3b-instruct-2507",
)

@entrypoint(store=store)  # 创建一个 LangGraph 工作流
async def chat(message: str): 
    response = model_qwen.invoke(message) 

    # memory_manager 从对话历史中提取记忆
    # 我们将以 OpenAI 的消息格式提供它
    to_process = {"messages": [{"role": "user", "content": message}] + [response]} 
    await memory_manager.ainvoke(to_process)  
    return response.content 

# 正常运行对话
response = await chat.ainvoke( 
    "记住我是张熙浚", 
) 
print(response) 

print(store.search(("memories",)))

git clone https://github.com/a2aproject/a2a-samples.git -b main --depth 1
cd a2a-samples

skill = AgentSkill(
    id='hello_world',
    name='Returns hello world',
    description='just returns hello world',
    tags=['hello world'],
    examples=['hi', 'hello world'],
)

# This will be the public-facing agent card
public_agent_card = AgentCard(
    name='Hello World Agent',
    description='Just a hello world agent',
    url='http://localhost:9999/',
    version='1.0.0',
 	# 默认输入模式：Agent 能够接收的输入类型列表，这里仅支持纯文本
    default_input_modes=['text'],
    # 默认输出模式：Agent 能够产生的输出类型列表，这里仅返回纯文本
    default_output_modes=['text'],
    # 能力声明：告知调用方 Agent 支持的能力，例如是否支持流式输出（streaming）
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],  # Only the basic skill for the public card
    supports_authenticated_extended_card=True,
)

import uvicorn

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import (
    HelloWorldAgentExecutor,  # type: ignore[import-untyped]
)


if __name__ == '__main__':
    skill = AgentSkill(
        id='hello_world',
        name='返回 hello world',
        description='简单地返回 hello world',
        tags=['hello world'],
        examples=['hi', 'hello world'],
    )

    extended_skill = AgentSkill(
        id='super_hello_world',
        name='返回 SUPER Hello World',
        description='仅限已认证用户使用的更热情的问候。',
        tags=['hello world', 'super', 'extended'],
        examples=['super hi', 'give me a super hello'],
    )

    # 这是面向公众的 Agent 卡片
    public_agent_card = AgentCard(
        name='Hello World Agent',
        description='只是一个 hello world 代理',
        url='http://localhost:9999/',
        version='1.0.0',
        default_input_modes=['text'],
        default_output_modes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],  # 公开卡片仅包含基础技能
        supports_authenticated_extended_card=True,
    )

    # 这是已认证用户的扩展 Agent 卡片
    # 额外包含 'extended_skill'
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Hello World Agent - Extended Edition',  # 使用不同名称以便区分
            'description': '面向已认证用户的完整功能 hello world 代理。',
            'version': '1.0.1',  # 甚至可以是不同的版本
            # capabilities 及其他字段（如 url、default_input_modes、default_output_modes、
            # supports_authenticated_extended_card）均从 public_agent_card 继承，
            # 除非在此处另行指定。
            'skills': [
                skill,
                extended_skill,
            ],  # 扩展卡片包含两个技能
        }
    )

    request_handler = DefaultRequestHandler(
        agent_executor=HelloWorldAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )

    # 使用 uvicorn 启动服务，监听 0.0.0.0:9999
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)