@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str)->str:
    return rag.rag_summarize(query)

@tool(description="从向量存储中检索参考资料")
def rag_summarize(query: str)->str:
    return rag.rag_summarize(query)

from langchain.tools import BaseTool

class MyCustomTool(BaseTool): # 必须注册name description 必须覆写_run
    # 
    @property
    def name(self):
        return "The unique name of the tool that clearly communicates its purpose"  

    @property
    def description(self):
        return "Used to tell the model how/when/why to use the tool"

    
    def _run(self, query: str):
        return "result"

from langchain.tools import BaseTool

class MyCustomTool(BaseTool): # 必须注册name description 必须覆写_run
    # 
    @property
    def name(self):
        return "The unique name of the tool that clearly communicates its purpose"  

    @property
    def description(self):
        return "Used to tell the model how/when/why to use the tool"

    
    def _run(self, query: str):
        return "result"

class Runnable(ABC, Generic[Input, Output]):
    """A unit of work that can be invoked, batched, streamed, transformed and composed.

    Key Methods
    ===========

    - `invoke`/`ainvoke`: Transforms a single input into an output.
    - `batch`/`abatch`: Efficiently transforms multiple inputs into outputs.
    - `stream`/`astream`: Streams output from a single input as it's produced.
    - `astream_log`: Streams output and selected intermediate results from an
        input.

    Built-in optimizations:

    - **Batch**: By default, batch runs invoke() in parallel using a thread pool
        executor. Override to optimize batching.

    - **Async**: Methods with `'a'` prefix are asynchronous. By default, they execute
        the sync counterpart using asyncio's thread pool.
        Override for native async.

    All methods accept an optional config argument, which can be used to configure
    execution, add tags and metadata for tracing and debugging etc.

    Runnables expose schematic information about their input, output and config via
    the `input_schema` property, the `output_schema` property and `config_schema`
    method.

    Composition
    ===========

    Runnable objects can be composed together to create chains in a declarative way.

    Any chain constructed this way will automatically have sync, async, batch, and
    streaming support.

    The main composition primitives are `RunnableSequence` and `RunnableParallel`.

    **`RunnableSequence`** invokes a series of runnables sequentially, with
    one Runnable's output serving as the next's input. Construct using
    the `|` operator or by passing a list of runnables to `RunnableSequence`.

    **`RunnableParallel`** invokes runnables concurrently, providing the same input
    to each. Construct it using a dict literal within a sequence or by passing a
    dict to `RunnableParallel`.

class Runnable(ABC, Generic[Input, Output]):
    """A unit of work that can be invoked, batched, streamed, transformed and composed.

    Key Methods
    ===========

    - `invoke`/`ainvoke`: Transforms a single input into an output.
    - `batch`/`abatch`: Efficiently transforms multiple inputs into outputs.
    - `stream`/`astream`: Streams output from a single input as it's produced.
    - `astream_log`: Streams output and selected intermediate results from an
        input.

    Built-in optimizations:

    - **Batch**: By default, batch runs invoke() in parallel using a thread pool
        executor. Override to optimize batching.

    - **Async**: Methods with `'a'` prefix are asynchronous. By default, they execute
        the sync counterpart using asyncio's thread pool.
        Override for native async.

    All methods accept an optional config argument, which can be used to configure
    execution, add tags and metadata for tracing and debugging etc.

    Runnables expose schematic information about their input, output and config via
    the `input_schema` property, the `output_schema` property and `config_schema`
    method.

    Composition
    ===========

    Runnable objects can be composed together to create chains in a declarative way.

    Any chain constructed this way will automatically have sync, async, batch, and
    streaming support.

    The main composition primitives are `RunnableSequence` and `RunnableParallel`.

    **`RunnableSequence`** invokes a series of runnables sequentially, with
    one Runnable's output serving as the next's input. Construct using
    the `|` operator or by passing a list of runnables to `RunnableSequence`.

    **`RunnableParallel`** invokes runnables concurrently, providing the same input
    to each. Construct it using a dict literal within a sequence or by passing a
    dict to `RunnableParallel`.

chain=({"input":RunnablePassThrough(),"context":retriever|format_func}|prompt|model|StrOutputParser())

chain=({"input":RunnablePassThrough(),"context":retriever|format_func}|prompt|model|StrOutputParser())

for chunk in agent.stream(
    {"messages":[{"role":"user","content":"..."}]}
    stream_mode = "values"
):
    latest_message = chunk["message"][-1]
    if latest_message.content:
        print(type(latest_message.__name__),latest_message.content)
    try:
        if latest_message.tool_calls:
            print(f"工具调用：{[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        pass

for chunk in agent.stream(
    {"messages":[{"role":"user","content":"..."}]}
    stream_mode = "values"
):
    latest_message = chunk["message"][-1]
    if latest_message.content:
        print(type(latest_message.__name__),latest_message.content)
    try:
        if latest_message.tool_calls:
            print(f"工具调用：{[tc['name'] for tc in latest_message.tool_calls]}")
    except AttributeError as e:
        pass

@before_model
def log_before_model(
    state:AgentState,  # Agent状态记录
    runtime:Runtime,    #执行过程中上下文信息
):
    logger.info(f"[log_before_model]即将调用模型，带有{len(state['messages'])}个消息")
    logger.debug(f"[log_before_model]{type(state['messages'][-1]).__name__}|{state['messages'][-1].content.strip()}")

    return None

@before_model
def log_before_model(
    state:AgentState,  # Agent状态记录
    runtime:Runtime,    #执行过程中上下文信息
):
    logger.info(f"[log_before_model]即将调用模型，带有{len(state['messages'])}个消息")
    logger.debug(f"[log_before_model]{type(state['messages'][-1]).__name__}|{state['messages'][-1].content.strip()}")

    return None

@wrap_tool_call
def monitor_tool(
    # 请求的数据结构
    request: ToolCallRequest,
    #执行的函数本身(Callable),它接受ToolCallRequest作为输入，并返回ToolMessage或Command
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
)-> ToolMessage | Command:
    
    logger.info(f"[tool monitor]执行工具：{request.tool_call['name']}")
    logger.info(f"[tool monitor]执行工具：{request.tool_call['args']}")
    try:
        result= handler(request)
        logger.info(f"[tool monitor]工具{request.tool_call['name']}执行完毕")


        if request.tool_call['name']=="fill_context_for_report":
            request.runtime.context["report"]=True
            logger.info(f"[tool monitor]工具{request.tool_call['name']}触发了报告场景的上下文填充")
        return result
    except Exception as e:
        logger.error(f"[tool monitor]工具{request.tool_call['name']}执行失败:{e}")
        raise e

@wrap_tool_call
def monitor_tool(
    # 请求的数据结构
    request: ToolCallRequest,
    #执行的函数本身(Callable),它接受ToolCallRequest作为输入，并返回ToolMessage或Command
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
)-> ToolMessage | Command:
    
    logger.info(f"[tool monitor]执行工具：{request.tool_call['name']}")
    logger.info(f"[tool monitor]执行工具：{request.tool_call['args']}")
    try:
        result= handler(request)
        logger.info(f"[tool monitor]工具{request.tool_call['name']}执行完毕")


        if request.tool_call['name']=="fill_context_for_report":
            request.runtime.context["report"]=True
            logger.info(f"[tool monitor]工具{request.tool_call['name']}触发了报告场景的上下文填充")
        return result
    except Exception as e:
        logger.error(f"[tool monitor]工具{request.tool_call['name']}执行失败:{e}")
        raise e

@dynamic_prompt  #每一次生成提示词之前调用
def report_prompt_switch(request:ModelRequest):
    is_report = request.runtime.context.get("report",False)
    if is_report:
        return load_report_prompt()
    
    return load_system_prompt()

@dynamic_prompt  #每一次生成提示词之前调用
def report_prompt_switch(request:ModelRequest):
    is_report = request.runtime.context.get("report",False)
    if is_report:
        return load_report_prompt()
    
    return load_system_prompt()