pip install git+https://github.com/emcie-co/parlant@develop

WARNING: Ignoring invalid distribution ~umpy (D:\ProgramData\Anaconda3\Lib\site-packages)Collecting git+https://github.com/emcie-co/parlant@develop  Cloning https://github.com/emcie-co/parlant (to revision develop) to c:\users\liang\appdata\local\temp\pip-req-build-jtzrkmza  Running command git clone --filter=blob:none --quiet https://github.com/emcie-co/parlant 'C:\Users\liang\AppData\Local\Temp\pip-req-build-jtzrkmza'  Host key verification failed.  fatal: Could not read from remote repository.
  Please make sure you have the correct access rights  and the repository exists.  error: subprocess-exited-with-error
  × git clone --filter=blob:none --quiet https://github.com/emcie-co/parlant 'C:\Users\liang\AppData\Local\Temp\pip-req-build-jtzrkmza' did not run successfully.  │ exit code: 128  ╰─> See above for output.
  note: This error originates from a subprocess, and is likely not a problem with pip.error: subprocess-exited-with-error
× git clone --filter=blob:none --quiet https://github.com/emcie-co/parlant 'C:\Users\liang\AppData\Local\Temp\pip-req-build-jtzrkmza' did not run successfully.│ exit code: 128╰─> See above for output.
note: This error originates from a subprocess, and is likely not a problem with pip.

# Copyright 2025 Emcie Co Ltd.## Licensed under the Apache License, Version 2.0 (the "License");# you may not use this file except in compliance with the License.# You may obtain a copy of the License at##     http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.# Maintainer: Ji Qing <jiqing19861123@163.com>from __future__ import annotationsimport timefrom openai import (    APIConnectionError,    APIResponseValidationError,    APITimeoutError,    AsyncClient,    ConflictError,    InternalServerError,    RateLimitError,)from typing import Any, Callable, Mappingfrom typing_extensions import overrideimport jsonimport jsonfinder  # type: ignoreimport osfrom pydantic import ValidationErrorimport tiktokenfrom parlant.adapters.nlp.common import normalize_json_outputfrom parlant.core.engines.alpha.prompt_builder import PromptBuilderfrom parlant.core.loggers import Loggerfrom parlant.core.nlp.policies import policy, retryfrom parlant.core.nlp.tokenization import EstimatingTokenizerfrom parlant.core.nlp.service import NLPServicefrom parlant.core.nlp.embedding import Embedder, EmbeddingResultfrom parlant.core.nlp.generation import (    T,    SchematicGenerator,    SchematicGenerationResult,)from parlant.core.nlp.generation_info import GenerationInfo, UsageInfofrom parlant.core.nlp.moderation import (    ModerationService,    NoModeration,)RATE_LIMIT_ERROR_MESSAGE = """\Qwen API rate limit exceeded. Possible reasons:1. Your account may have insufficient API credits.2. You may be using a free-tier account with limited request capacity.3. You might have exceeded the requests-per-minute limit for your account.Recommended actions:- Check your Qwen account balance and billing status.- Review your API usage limits in Qwen's dashboard.- For more details on rate limits and usage tiers, visit:    https://docs.bigmodel.cn/cn/faq/api-code"""class QwenEstimatingTokenizer(EstimatingTokenizer):    def __init__(self, model_name: str) -> None:        self.model_name = model_name        self.encoding = tiktoken.encoding_for_model("gpt-4o-2024-08-06")    @override    async def estimate_token_count(self, prompt: str) -> int:        tokens = self.encoding.encode(prompt)        return len(tokens)class QwenEmbedder(Embedder):    supported_arguments = ["dimensions"]    def __init__(self, model_name: str, logger: Logger) -> None:        self.model_name = model_name        self._logger = logger        self._client = AsyncClient(            base_url=os.environ.get(                "BASE_URL", "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"            ),            api_key=os.environ.get("DASHSCOPE_API_KEY", ""),        )        self._tokenizer = QwenEstimatingTokenizer(model_name=self.model_name)    @property    @override    def id(self) -> str:        return f"qwen/{self.model_name}"    @property    @override    def tokenizer(self) -> QwenEstimatingTokenizer:        return self._tokenizer    @policy(        [            retry(                exceptions=(                    APIConnectionError,                    APITimeoutError,                    ConflictError,                    RateLimitError,                    APIResponseValidationError,                ),            ),            retry(InternalServerError, max_exceptions=2, wait_times=(1.0, 5.0)),        ]    )    @override    async def embed(        self,        texts: list[str],        hints: Mapping[str, Any] = {},    ) -> EmbeddingResult:        filtered_hints = {k: v for k, v in hints.items() if k in self.supported_arguments}        try:            response = await self._client.embeddings.create(                model=self.model_name,                input=texts,                **filtered_hints,            )        except RateLimitError:            self._logger.error(RATE_LIMIT_ERROR_MESSAGE)            raise        vectors = [data_point.embedding for data_point in response.data]        return EmbeddingResult(vectors=vectors)class QwenTextEmbedding_V4(QwenEmbedder):    def __init__(self, logger: Logger) -> None:        super().__init__(model_name="qwen3-embedding-8b", logger=logger)                # Override the client to use the embedding-specific endpoint        self._client = AsyncClient(            base_url=os.environ.get(                "EMBEDDING_BASE_URL", "http://你的IP:你的端口/v1"            ),            api_key=os.environ.get("DASHSCOPE_API_KEY", ""),        )    @property    @override    def max_tokens(self) -> int:        return 8192    @property    def dimensions(self) -> int:        return 4096class QwenSchematicGenerator(SchematicGenerator[T]):    supported_qwen_params = ["temperature", "max_tokens"]    def __init__(        self,        model_name: str,        logger: Logger,    ) -> None:        self.model_name = model_name        self._logger = logger        self._client = AsyncClient(            base_url=os.environ.get(                "BASE_URL", "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"            ),            api_key=os.environ["DASHSCOPE_API_KEY"],        )        self._tokenizer = QwenEstimatingTokenizer(model_name=self.model_name)    @property    @override    def id(self) -> str:        return f"Qwen/{self.model_name}"    @property    @override    def tokenizer(self) -> QwenEstimatingTokenizer:        return self._tokenizer    @policy(        [            retry(                exceptions=(                    APIConnectionError,                    APITimeoutError,                    ConflictError,                    RateLimitError,                    APIResponseValidationError,                ),            ),            retry(InternalServerError, max_exceptions=2, wait_times=(1.0, 5.0)),        ]    )    @override    async def generate(        self,        prompt: str | PromptBuilder,        hints: Mapping[str, Any] = {},    ) -> SchematicGenerationResult[T]:        with self._logger.operation(f"Qwen LLM Request ({self.schema.__name__})"):            return await self._do_generate(prompt, hints)    async def _do_generate(        self,        prompt: str | PromptBuilder,        hints: Mapping[str, Any] = {},    ) -> SchematicGenerationResult[T]:        if isinstance(prompt, PromptBuilder):            prompt = prompt.build()        qwen_api_arguments = {k: v for k, v in hints.items() if k in self.supported_qwen_params}        t_start = time.time()        response = await self._client.chat.completions.create(            messages=[{"role": "user", "content": prompt}],            model=self.model_name,            max_tokens=8 * 1024,            response_format={"type": "json_object"},            **qwen_api_arguments,        )        t_end = time.time()        if response.usage:            self._logger.trace(response.usage.model_dump_json(indent=2))        raw_content = response.choices[0].message.content or "{}"        try:            json_content = json.loads(normalize_json_output(raw_content))            # Fix common schema issues: convert conditions from list to string            if isinstance(json_content, dict) and 'actions' in json_content:                for action in json_content['actions']:                    if isinstance(action.get('conditions'), list) and action['conditions']:                        action['conditions'] = action['conditions'][0]                    elif isinstance(action.get('conditions'), list) and not action['conditions']:                        action['conditions'] = ""        except json.JSONDecodeError:            self._logger.warning(f"Invalid JSON returned by {self.model_name}:\n{raw_content})")            json_content = jsonfinder.only_json(raw_content)[2]            self._logger.warning("Found JSON content within model response; continuing...")            # Apply the same fix to extracted JSON            if isinstance(json_content, dict) and 'actions' in json_content:                for action in json_content['actions']:                    if isinstance(action.get('conditions'), list) and action['conditions']:                        action['conditions'] = action['conditions'][0]                    elif isinstance(action.get('conditions'), list) and not action['conditions']:                        action['conditions'] = ""        try:            content = self.schema.model_validate(json_content)            assert response.usage            return SchematicGenerationResult(                content=content,                info=GenerationInfo(                    schema_name=self.schema.__name__,                    model=self.id,                    duration=(t_end - t_start),                    usage=UsageInfo(                        input_tokens=response.usage.prompt_tokens,                        output_tokens=response.usage.completion_tokens,                        extra={                            "cached_input_tokens": getattr(                                response,                                "usage.prompt_cache_hit_tokens",                                0,                            )                        },                    ),                ),            )        except ValidationError:            self._logger.error(                f"JSON content returned by {self.model_name} does not match expected schema:\n{raw_content}"            )            raiseclass Qwen_MAX(QwenSchematicGenerator[T]):    def __init__(self, logger: Logger) -> None:        super().__init__(model_name="qwen-max", logger=logger)    @property    @override    def max_tokens(self) -> int:        return 32 * 1024class Qwen_Plus(QwenSchematicGenerator[T]):    def __init__(self, logger: Logger) -> None:        super().__init__(model_name="qwen-plus", logger=logger)    @property    @override    def max_tokens(self) -> int:        return 128 * 1024class Qwen_2_5_72b(QwenSchematicGenerator[T]):    def __init__(self, logger: Logger) -> None:        super().__init__(model_name="qwen2.5-72b-instruct", logger=logger)    @property    @override    def max_tokens(self) -> int:        return 128 * 1024class Qwen_3_30b_A3B(QwenSchematicGenerator[T]):    def __init__(self, logger: Logger) -> None:        super().__init__(model_name="qwen3-30b-a3b-instruct", logger=logger)    @property    @override    def max_tokens(self) -> int:        return 128 * 1024class QwenService(NLPService):    @staticmethod    def verify_environment() -> str | None:        """Returns an error message if the environment is not set up correctly."""        if not os.environ.get("DASHSCOPE_API_KEY"):            return """\You're using the Qwen NLP service, but DASHSCOPE_API_KEY is not set.Please set DASHSCOPE_API_KEY in your environment before running Parlant."""        return None    def __init__(        self,        logger: Logger,    ) -> None:        self._logger = logger        self._logger.info("Initialized QwenService")        self.model_name = os.environ.get("QWEN_MODEL", "qwen-plus")        self._logger.info(f"Qwen model name: {self.model_name}")    def _get_specialized_generator_class(        self,        model_name: str,        t: type[T],    ) -> Callable[..., QwenSchematicGenerator[T]] | None:        """        Returns the specialized generator class for known models        """        model_mapping: dict[str, type[QwenSchematicGenerator[T]]] = {            "qwen-max": Qwen_MAX[t],  # type: ignore            "qwen-plus": Qwen_Plus[t],  # type: ignore            "qwen2.5-72b-instruct": Qwen_2_5_72b[t],  # type: ignore            "qwen3-30b-a3b-instruct": Qwen_3_30b_A3B[t],  # type: ignore        }        if generator_class := model_mapping.get(model_name):            return generator_class        else:            return None    @override    async def get_schematic_generator(self, t: type[T]) -> QwenSchematicGenerator[T]:        qwen_generator = self._get_specialized_generator_class(self.model_name, t)        assert qwen_generator is not None, f"Unsupported Qwen model: {self.model_name}"        return qwen_generator(self._logger)    @override    async def get_embedder(self) -> Embedder:        return QwenTextEmbedding_V4(logger=self._logger)    @override    async def get_moderation_service(self) -> ModerationService:        return NoModeration()

# healthcare.pyimport parlant.sdk as pimport asyncioimport osfrom datetime import datetime@p.toolasync def get_insurance_providers(context: p.ToolContext) -> p.ToolResult:    return p.ToolResult(["大型保险公司", "顶点保险公司"])@p.toolasync def get_upcoming_slots(context: p.ToolContext) -> p.ToolResult:    # 模拟从数据库或API获取可用时间段    return p.ToolResult(data=["周一上午10点", "周二下午2点", "周三下午1点"])@p.toolasync def get_later_slots(context: p.ToolContext) -> p.ToolResult:    # 模拟获取后续可用时间段    return p.ToolResult(data=["11月3日上午11:30", "11月12日下午3点"])@p.toolasync def schedule_appointment(context: p.ToolContext, datetime: datetime) -> p.ToolResult:    # 模拟预约安排    return p.ToolResult(data=f"预约已安排在 {datetime}")@p.toolasync def get_lab_results(context: p.ToolContext) -> p.ToolResult:    # 模拟从数据库或API获取化验结果，    # 使用上下文中的客户ID。    lab_results = {        "report": "所有检测结果都在正常范围内",        "prognosis": "患者健康状况良好！",    }    return p.ToolResult(        data={            "report": lab_results["report"],            "prognosis": lab_results["prognosis"],        }    )async def add_domain_glossary(agent: p.Agent) -> None:    await agent.create_term(        name="办公室电话号码",        description="我们办公室的电话号码，+1-234-567-8900",    )    await agent.create_term(        name="办公时间",        description="办公时间为周一至周五，上午9点到下午5点",    )    await agent.create_term(        name="查尔斯·泽维尔",        synonyms=["X教授"],        description="专门从事神经科工作的医生，周一和周二坐诊。",    )    # 如需要，在此处添加其他特定术语和定义...# <<添加此函数>>async def create_scheduling_journey(server: p.Server, agent: p.Agent) -> p.Journey:    # 创建旅程    journey = await agent.create_journey(        title="预约安排",        description="帮助患者找到适合的预约时间。",        conditions=["患者想要预约"],    )    # 首先，确定预约原因    t0 = await journey.initial_state.transition_to(chat_state="确定就诊原因")    # 将即将到来的预约时间段加载到上下文中    t1 = await t0.target.transition_to(tool_state=get_upcoming_slots)    # 询问哪个时间段适合他们    # 我们将根据患者的回复从此处有条件地转换    t2 = await t1.target.transition_to(        chat_state="列出可用时间并询问哪个时间段适合他们"    )    # 我们从患者选择时间的愉快路径开始    t3 = await t2.target.transition_to(        chat_state="在安排前与患者确认详细信息",        condition="患者选择了时间",    )    t4 = await t3.target.transition_to(        tool_state=schedule_appointment,        condition="患者确认了详细信息",    )    t5 = await t4.target.transition_to(chat_state="确认预约已安排")    await t5.target.transition_to(state=p.END_JOURNEY)    # 否则，如果他们说这些时间都不行，询问后续时间段    t6 = await t2.target.transition_to(        tool_state=get_later_slots,        condition="这些时间对患者都不合适",    )    t7 = await t6.target.transition_to(chat_state="列出后续时间并询问是否有合适的时间")    # 如果他们选择时间，转换回我们的愉快路径    await t7.target.transition_to(state=t3.target, condition="患者选择了时间")    # 否则，请他们致电办公室    t8 = await t7.target.transition_to(        chat_state="请患者致电办公室预约",        condition="这些时间对患者也都不合适",    )    await t8.target.transition_to(state=p.END_JOURNEY)    # 使用指导原则有意识地处理边缘情况    await journey.create_guideline(        condition="患者说他们的就诊很紧急",        action="告诉他们立即致电办公室",    )    return journeyasync def create_lab_results_journey(server: p.Server, agent: p.Agent) -> p.Journey:    # 创建旅程    journey = await agent.create_journey(        title="化验结果",        description="获取患者的化验结果并进行解释。",        conditions=["患者想要查看他们的化验结果"],    )    t0 = await journey.initial_state.transition_to(tool_state=get_lab_results)    await t0.target.transition_to(        chat_state="告诉患者结果尚未可用，请稍后再试",        condition="找不到化验结果",    )    await t0.target.transition_to(        chat_state="向患者解释化验结果 - 结果正常",        condition="化验结果良好 - 即无需担心",    )    await t0.target.transition_to(        chat_state="呈现结果并请他们致电办公室 "        "对结果进行澄清，因为你不是医生",        condition="化验结果不好 - 即患者健康有问题",    )    # 使用指导原则处理边缘情况...    await agent.create_guideline(        condition="患者催促你提供更多关于化验结果的结论",        action="明确告诉他们你无法帮助，他们应该致电办公室",    )    return journeyasync def main() -> None:    # 设置千问服务配置    os.environ['BASE_URL'] = 'http://您的IP:端口/v1'  # 私有化部署的千问模型地址    os.environ['EMBEDDING_BASE_URL'] = 'http://您的IP:端口/v1'  # 私有化部署的embedding模型地址    os.environ['DASHSCOPE_API_KEY'] = 'not-needed'  # 私有化部署不需要 API key    os.environ['QWEN_MODEL'] = 'qwen3-30b-a3b-instruct'  # 使用的模型名称        async with p.Server(nlp_service=p.NLPServices.qwen) as server:        agent = await server.create_agent(            name="医疗助手",            description="对患者富有同情心且能安抚情绪。",        )        await agent.create_canned_response("今天我能为您做什么？")        await add_domain_glossary(agent)        scheduling_journey = await create_scheduling_journey(server, agent)        lab_results_journey = await create_lab_results_journey(server, agent)        status_inquiry = await agent.create_observation(            "患者询问跟进就诊情况，但具体方式不明确",        )        # 使用此观察来区分两个旅程        await status_inquiry.disambiguate([scheduling_journey, lab_results_journey])        await agent.create_guideline(            condition="患者询问保险问题",            action="列出我们接受的保险提供商，并告诉他们致电办公室了解更多详情",            tools=[get_insurance_providers],        )        await agent.create_guideline(            condition="患者要求与人工代理通话",            action="请他们致电办公室，提供电话号码",        )        await agent.create_guideline(            condition="患者询问与我们医疗保健无关的事情",            action="礼貌地告诉您无法协助处理无关询问 - 不要回应他们的请求。",        )if __name__ == "__main__":    asyncio.run(main())

Traceback (most recent call last):  File "D:\ProgramData\anaconda3\Lib\site-packages\parlant\sdk.py", line 64, in <module>    from parlant.adapters.vector_db.transient import TransientVectorDatabase  File "D:\ProgramData\anaconda3\Lib\site-packages\parlant\adapters\vector_db\transient.py", line 50, in <module>    import nano_vectordb  # type: ignore    ^^^^^^^^^^^^^^^^^^^^  File "D:\ProgramData\anaconda3\Lib\site-packages\nano_vectordb\__init__.py", line 1, in <module>    from .dbs import NanoVectorDB, MultiTenantNanoVDB  File "D:\ProgramData\anaconda3\Lib\site-packages\nano_vectordb\dbs.py", line 18, in <module>    Data = TypedDict("Data", {"__id__": str, "__vector__": np.ndarray})                                                           ^^^^^^^^^^AttributeError: module 'numpy' has no attribute 'ndarray'

pip install "numpy<2.0"

[Evaluation(built-in:get_lab_results)] SingleToolBatch attempt 0 failed: ['Traceback (most recent call last):\n', '  File "D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\single_tool_batch.py", line 228, in _infer_calls_for_single_tool\n    ) = await self._evaluate_tool_calls(inference_output, candidate_descriptor)\n        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', '  File "D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\single_tool_batch.py", line 262, in _evaluate_tool_calls\n    descriptor, options = tool.parameters[evaluation.parameter_name]\n                          ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', "KeyError: 'None'\n"]

2025-09-29T02:57:26.747269Z [error    ] [R2mEaHAj0vh::process][ToolCaller] Processing context for session MkDEb63x0k failed2025-09-29T02:57:26.749230Z [error    ] [R2mEaHAj0vh::process][ToolCaller] Traceback (most recent call last):   File "D:\ProgramData\anaconda3\Lib\site-packages\parlant\core\engines\alpha\tool_calling\single_tool_batch.py", line 228, in _infer_calls_for_single_tool    ) = await self._evaluate_tool_calls(inference_output, candidate_descriptor)        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^   File "D:\ProgramData\anaconda3\Lib\site-packages\parlant\core\engines\alpha\tool_calling\single_tool_batch.py", line 262, in _evaluate_tool_calls    descriptor, options = tool.parameters[evaluation.parameter_name]                          ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^ KeyError: 'reason_for_visit'
 '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_event_generator.py", ' 'line 131, in generate_events\n' '    inference_result = await self._tool_caller.infer_tool_calls(\n' '                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', '  File ' '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\tool_caller.py", ' 'line 182, in infer_tool_calls\n' '    return await self._do_infer_tool_calls(context)\n' '           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', '  File ' '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\tool_caller.py", ' 'line 220, in _do_infer_tool_calls\n' '    batch_results = await async_utils.safe_gather(*batch_tasks)\n' '                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', '  File ' '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\async_utils.py", ' 'line 148, in safe_gather\n' '    return await asyncio.gather(\n' '           ^^^^^^^^^^^^^^^^^^^^^\n', '  File ' '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\single_tool_batch.py", ' 'line 138, in process\n' '    ) = await self._infer_calls_for_single_tool(\n' '        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n', '  File ' '"D:\\ProgramData\\anaconda3\\Lib\\site-packages\\parlant\\core\\engines\\alpha\\tool_calling\\single_tool_batch.py", ' 'line 239, in _infer_calls_for_single_tool\n' '    raise ToolCallBatchError() from last_generation_exception\n', 'parlant.core.engines.alpha.tool_calling.tool_caller.ToolCallBatchError: Tool ' 'Call Batch failed\n']

await agent.create_guideline(    condition="用户询问商品库存或 availability",    action="获取用户提及的商品ID，调用查库存工具，用模板告知库存状态",    tools=[check_stock]  # 工具只在这个条件下触发，不瞎调用)

# 1. 商品库存查询await agent.create_guideline(    condition="用户询问商品是否有货、库存数量或尺码 availability",    action="提取商品ID/名称，调用查库存工具，用库存模板回复",    tools=[check_stock])# 2. 订单退款咨询await agent.create_guideline(    condition="用户提及退款、退货或退款到账时间",    action="获取订单号，调用查退款状态工具，用退款模板回复",    tools=[check_refund_status])# 3. 商品推荐await agent.create_guideline(    condition="用户请求推荐商品，或提及品类+预算",    action="提取品类（如口红）和预算，调用推荐工具，用推荐模板回复",    tools=[recommend_products])

# 创建“商品咨询→下单”旅程journey = await agent.create_journey(    title="商品购买引导",    conditions=["用户提及商品、表达购买意向"],    description="从咨询到下单的灵活引导流程")# 状态1：商品咨询→查库存t1 = await journey.initial_state.transition_to(    tool_state=check_stock,  # 触发查库存工具    condition="用户问库存")# 状态2：库存有货→推荐尺码/搭配t2 = await t1.target.transition_to(    chat_state="询问用户需要的尺码，或推荐搭配商品",    condition="库存有货")# 状态3：用户确认购买→调用下单工具t3 = await t2.target.transition_to(    tool_state=create_order,  # 触发下单工具    condition="用户说‘我要下单’或‘买了’")# 状态4：下单成功→告知物流await t3.target.transition_to(    chat_state="用物流模板告知预计送达时间，询问是否需要其他帮助",    condition="下单成功")# 兜底：库存无货→推荐相似款await t1.target.transition_to(    tool_state=recommend_similar,  # 触发相似商品推荐    condition="库存无货")

# 1. 查库存工具@p.toolasync def check_stock(context: p.ToolContext, product_id: str) -> p.ToolResult:    # 从电商系统获取库存数据    stock_data = await ecommerce_db.get_stock(product_id)    return p.ToolResult(        data={            "name": stock_data["name"],            "sizes": stock_data["available_sizes"],            "delivery": "2-3天"        },        canned_responses=[            "您好，{{name}}当前有{{sizes}}码现货，预计{{delivery}}送达~"        ]    )# 2. 推荐商品工具@p.toolasync def recommend_products(context: p.ToolContext, category: str, budget: float) -> p.ToolResult:    products = await ecommerce_db.search(category=category, price_max=budget, in_stock=True)    return p.ToolResult(        data={"products": products},        canned_responses=[            "根据您的需求，推荐{{products.length}}款商品：{{#each products}}{{name}}（¥{{price}}）{{#unless @last}}、{{/unless}}{{/each}}，需要了解某款的详情吗？"        ]    )

👤 用户：你好，这款黑色连衣裙有M码吗？🤖 小电：稍等，我查一下库存…您好，黑色连衣裙当前有M和L码现货，预计2-3天送达~需要了解尺码表吗？👤 用户：不用，我直接买M码🤖 小电：好的，正在为您创建订单…订单创建成功！订单号是ORD20250929001，预计9月30日送达，收货地址是您上次的“XX小区1号楼”，对吗？👤 用户：对的，谢谢🤖 小电：不客气！后续有订单问题，随时找我哦~

# 同时创建3个Agentasync with p.Server() as server:    cs_agent = await server.create_agent(name="电商客服Agent")  # 处理咨询    after_sales_agent = await server.create_agent(name="售后Agent")  # 处理退款    logistics_agent = await server.create_agent(name="物流Agent")  # 查物流# 客服Agent处理不了的售后问题，自动转售后Agentawait cs_agent.create_guideline(    condition="用户问题涉及售后退款，且客服无法解决",    action="转接售后Agent，并同步对话历史",    tools=[transfer_to_agent(after_sales_agent)])