1121

一、什么是大模型应用

• 大模型应用的特点：
  • 强调模型能力和业务场景结合，解决实际的业务问题
  • 要求模型能力能够融入应用的业务逻辑实现
  • 模型能力是应用的业务逻辑实现中的重要环节，但也只是应用的业务逻辑实现中的一环
• 容易混淆的概念：
  • 大模型开发、大模型训练、大模型微调
  • 模型推理、MaaS

二、大模型应用的三个关键要素

三、模型能力在开发中的实际效果

!pip install OpenAI

云端API：

import ENV
from openai import OpenAI

# 创建一个 OpenAI 调用客户端，配置相关环境参数
client = OpenAI(
    api_key=ENV.DEEPSEEK_API_KEY,
    base_url=ENV.DEEPSEEK_URL
)
# 调用 ChatCompletion 接口
response = client.chat.completions.create(
    model=ENV.DEEPSEEK_MODEL,
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Translate 'Hello, world!' into French."}
    ],
    max_tokens=50,
    temperature=0.7
)

# 输出结果
print(response.choices[0].message.content.strip())

"Bonjour, le monde !"

本地模型服务：

from openai import OpenAI

client = OpenAI(
    api_key="nothing",
    base_url="http://localhost:11434/v1"
)
# 调用 ChatCompletion 接口
response = client.chat.completions.create(
    model="qwen2.5",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Translate 'Hello, world!' into French."}
    ],
    max_tokens=50,
    temperature=0.7
)

# 输出结果
print(response.choices[0].message.content.strip())

The translation of "Hello, world!" into French is "Salut, le monde!"

Please note that this is a common programming phrase and may not be the most formal or appropriate greeting in all contexts. If you're looking for a more formal

简单封装一下以备后续使用：

import ENV
from openai import OpenAI

def direct_request_model(user_input, model="deepseek"):
    if model=="qwen": 
        client = OpenAI(
            api_key="nothing",
            base_url="http://localhost:11434/v1",
        )
    else:
        client = OpenAI(
            api_key=ENV.DEEPSEEK_API_KEY,
            base_url=ENV.DEEPSEEK_URL,
        )

    response = client.chat.completions.create(
        model="qwen2.5" if model == "qwen" else ENV.DEEPSEEK_MODEL,
        messages=[
            {"role": "user", "content": user_input}
        ],
        max_tokens=4096,
        temperature=0.7
    )
    return response.choices[0].message.content.strip()

print(direct_request_model("请给我输出5个单词和3个句子", "qwen"))

五个单词：梦想、勇气、坚持、成功、希望。

三个句子：我相信努力一定会带来成功。每个人都有追求梦想的权利。无论遇到多大困难，都不能放弃希望。

在这里，我们就遇到了第一个工程开发场景里的难题：这些自然语言里的关键信息，怎么用到代码开发里？

四、结构性数据输出——模型能力在工程开发中的第一道槛

print(
    direct_request_model("""
请给我输出5个单词和3个句子
输出时请使用JSON格式
第一个字段名是'words'，里面用列表的方式放入5个str的单词元素
第二个字段名是'sentences'，里面用列表的方式放入3个str的句子元素
请务必只输出JSON字符串，不要有额外的说明和修饰
""", "qwen")
)

```json
{"words":["happy","dog","sunshine","run","cake"],"sentences":["I feel happy when I see my dog.","The sunshine makes the day perfect.", "Let's go for a run after breakfast."]}
```

# 版本依赖要求>=3.4.0.5
!pip install -U Agently

import Agently

client = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", "http://localhost:11434/v1")
        .set_settings("model.OAIClient.options", { "model": "qwen2.5" })
)

result = (
    client
        .instruct("输出5个单词和3个句子")
        .output({
            "intention": ("'任务', '闲聊', '知识问答'", "从给定选项中选择{instruct}的指令类型"),
            "result": {
                "words": [("str", )],
                "sentences": [("str", )],
            },
            "extra_task": ([{
                "chosen_word": ("str", ),
                "sentence": ("str", ),
            }], "从{result.words}中选择2个单词，每个单词各造一个句子"),
        })
        .start()
)

print("意图:", result["intention"])
print("结果:", result["result"])
print("额外任务:", result["extra_task"])

意图: 任务
结果: {'words': ['天空', '自由', '梦想', '星光', '探索'], 'sentences': ['追逐梦想如同仰望星空。', '每个人心中都有无限的自由。', '探索未知，才能见到更美的世界。']}
额外任务: [{'chosen_word': '梦想', 'sentence': '追梦的路上从来都不孤单。'}, {'chosen_word': '自由', 'sentence': '心灵的自由比任何物质都珍贵。'}]

五、一个常见的错误认知大模型应用的公式

模型服务 + PROMPT = 期望结果

在这个公式下的应用团队心理

• 模型服务部分
  • 模型能力做得强点智能点，就能解决我的问题
    • -> XX模型太弱了，还是GPT-4好用，还得是国外大厂的顶尖模型厉害
  • 模型厂商应该贴近我们应用层，解决我们应用层的痛点
    • -> JSON Mode、Function Calling都没有，这个模型不太行（一票否决绝大部分开源模型）
  • 便宜模型没好货，能搜索的、带知识库的模型又太贵了
    • -> 害，现在这个阶段，成本根本降不下来，ROI都打不正，做什么应用
• PROMPT部分
  • 模型能力调用就是跟模型说清楚话，所以找一个会和模型说话的人很重要
    • -> 提示词工程师，主要就是写提示词，所以人人都可以是提示词“工程师”
  • 把模型接口包好了，就是PROMPT的事情了，中间没什么业务层，这一层很薄的
    • -> 什么大模型应用，不就是给套壳嘛，我们工程团队搞搞并发、稳定性的事情就好了，业务的事情让产品经理、运营来做，接口都给你包好了，你去写提示词啊！
  • One/Few Shots、CoT、ReAct方法论一套一套的，但和我工程有什么关系呢？
    • -> 你们算法 / 提示词工程师去研究吧，PROMPT讲清楚了，业务就讲清楚了

然而，真的是这样吗？

• 在不是问答机器人的场景里，模型输出的，和我们想让用户看到的，是一样的吗？
• 从另一个角度想，模型的输出，只是直接给末端用户使用的吗？有没有可能，有些输出是给模型自己看的，有些输出是其他工程模块消费的？
• 以及，有多少业务场景，是一次请求就能处理好的？

六、一个稍有质量的客服机器人，也有值得编排的业务逻辑

import ENV
import Agently

agent = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", ENV.DEEPSEEK_URL)
        .set_settings("model.OAIClient.auth", { "api_key": ENV.DEEPSEEK_API_KEY })
        .set_settings("model.OAIClient.options", { "model": ENV.DEEPSEEK_MODEL })
)

workflow = Agently.Workflow()

@workflow.chunk()
def user_input(inputs, storage):
    storage.set("user_input", input("[请输入您的要求]: "))
    return

@workflow.chunk()
def judge_intent_and_quick_reply(inputs, storage):
    result = (
        agent
            .input(storage.get("user_input"))
            .output({
                "user_intent": ("闲聊 | 售后问题 | 其他", "判断用户提交的{input}内容属于给定选项中的哪一种"),
                "quick_reply": (
                    "str",
"""如果{user_intent}=='闲聊'，那么直接给出你的回应；
如果{user_intent}=='售后问题'，那么请用合适的方式告诉用户你已经明白用户的诉求，安抚客户情绪并请稍等你去看看应该如何处理；
如果{user_intent}=='其他'，此项输出null""")
            })
            .start()
    )
    storage.set("reply", result["quick_reply"])
    return result["user_intent"]

@workflow.chunk()
def generate_after_sales_reply(inputs, storage):
    storage.set("reply", (
        agent
            .input(storage.get("user_input"))
            .instruct(
"""请根据{input}的要求，以一个专业客户服务人员的角色给出回复，遵循如下模板进行回复：
亲爱的客户，感谢您的耐心等待。
我理解您希望{{复述客户的要求}}，是因为{{复述客户要求提出要求的理由}}，您的心情一定非常{{阐述你对客户心情/感受的理解}}。
{{给出对客户当前心情的抚慰性话语}}。
我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。
"""
)
            .start()
    ))
    return

@workflow.chunk()
def generate_other_topic_reply(inputs, storage):
    storage.set("reply", "我们好像不应该聊这个，还是回到您的问题或诉求上来吧。")
    return

@workflow.chunk_class()
def reply(inputs, storage):
    print("[回复]: ", storage.get("reply"))
    return

(
    workflow
        .connect_to("user_input")
        .connect_to("judge_intent_and_quick_reply")
        .if_condition(lambda return_value, storage: return_value=="闲聊")
            .connect_to("@reply")
            .connect_to("END")
        .elif_condition(lambda return_value, storage: return_value=="售后问题")
            .connect_to("@reply")
            .connect_to("generate_after_sales_reply")
            .connect_to("@reply")
            .connect_to("END")
        .else_condition()
            .connect_to("generate_other_topic_reply")
            .connect_to("@reply")
            .connect_to("END")
)

print(workflow.draw())

%%{ init: { 'flowchart': { 'curve': 'linear' }, 'theme': 'neutral' } }%%
%% Rendered By Agently %%
flowchart LR
classDef chunk_style fill:#fbfcdb,stroke:#666,stroke-width:1px,color:#333;
classDef condition_chunk_style fill:#ECECFF,stroke:#9370DB,stroke-width:1px,color:#333;
classDef loop_style fill:#f5f7fa,stroke:#666,stroke-width:1px,color:#333,stroke-dasharray: 5 5
    START("START"):::chunk_style -.-> |"#42; -->-- default"| user_input("user_input"):::chunk_style
    user_input("user_input"):::chunk_style -.-> |"#42; -->-- default"| judge_intent_and_quick_reply("judge_intent_and_quick_reply"):::chunk_style
    judge_intent_and_quick_reply("judge_intent_and_quick_reply"):::chunk_style -.-> |"#42; -->-- default"| 498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::6{{"Condition"}}:::condition_chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::6{{"Condition"}}:::condition_chunk_style -.-> |"#42; -- ◇ -- default"| 498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::7("@reply"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::7("@reply"):::chunk_style -.-> |"#42; -->-- default"| END("END"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::6{{"Condition"}}:::condition_chunk_style -.-> |"#42; -- ◇ -- default"| 498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::8("@reply"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::8("@reply"):::chunk_style -.-> |"#42; -->-- default"| generate_after_sales_reply("generate_after_sales_reply"):::chunk_style
    generate_after_sales_reply("generate_after_sales_reply"):::chunk_style -.-> |"#42; -->-- default"| 498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::9("@reply"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::9("@reply"):::chunk_style -.-> |"#42; -->-- default"| END("END"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::6{{"Condition"}}:::condition_chunk_style -.-> |"#42; -- ◇ -- default"| generate_other_topic_reply("generate_other_topic_reply"):::chunk_style
    generate_other_topic_reply("generate_other_topic_reply"):::chunk_style -.-> |"#42; -->-- default"| 498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::10("@reply"):::chunk_style
    498a5c09-079a-50a3-84f1-9c6845703f0c::chunk::10("@reply"):::chunk_style -.-> |"#42; -->-- default"| END("END"):::chunk_style

workflow.start()

[回复]:  非常抱歉听到您对蛋糕的体验不佳。我理解您的感受，请稍等，我去查看一下如何为您处理这个问题。
[回复]:  亲爱的客户，感谢您的耐心等待。
我理解您希望对昨天购买的蛋糕进行反馈，是因为蛋糕的味道让您感到不太满意，您的心情一定非常失望。
我们非常重视您的感受，并对此给您带来的不便深表歉意。
我们会尽快和相关人员沟通，并尽量进行满足。请留下您的联系方式以方便我们尽快处理后与您联系。





{'default': None}

七、所以这是个Agent？还是个Workflow？

点击展开

Agentic Workflow

• 更侧重以完成特定任务为目标
• 流程编排设计基于业务流程规划
• 模型输出能力只是业务流程中的一项可用能力
• 更重视使用模型能力处理业务中与模型能力适配的任务环节
• 其他系统模块也会处理与之能力适配的任务环节
• 强业务逻辑导向

点击展开

点击展开

Automatic Agent 通用自动规划智能体

• 以完成泛化任务为目标
• 更重视任务处理元方法的合理性
• 追求结构完整性和自举可能性
• 模型能力是驱动整个结构运转的核心
• 各个模块为自动化处理泛化任务而设计
• 强内部结构自洽自举导向

八、使用Agently框架，13行代码创建一个“AI搜索智能体”

"""创建Agent实例"""
import Agently
agent = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", ENV.DEEPSEEK_URL)
        .set_settings("model.OAIClient.auth", { "api_key": ENV.DEEPSEEK_API_KEY })
        .set_settings("model.OAIClient.options", { "model": ENV.DEEPSEEK_MODEL })
)
"""设定Agent实例可调用的工具"""
agent.use_public_tools(["search"])
"""发起请求"""
print(agent.input("目前英伟达的股价是多少").start())  

I don't know the current stock price of NVIDIA because the provided information indicates a rate limit error. You can try checking the latest stock price on a financial news website or a stock market platform.

九、能不能将工具调用过程白盒化

• 内置的工具不能用怎么办？
• ReAct决策过程黑盒不透明，我没有安全感
• ReAct不行！我有更好的决策方案！
• 我有一个在我的业务场景里更适合的处理流程，能固化到Agent上吗？

自定义工具

import ENV
import requests
import re
from bs4 import BeautifulSoup

# 搜索工具
def search(keywords:list):
    payload = json.dumps({
        "q": ' '.join(keywords),
    })
    headers = {
        'X-API-KEY': ENV.SERP_API_KEY,
        'Content-Type': 'application/json'
    }
    response = requests.request("POST", "https://google.serper.dev/search", headers=headers, data=payload)
    return response.text

# 浏览工具
def browse(url: str):
    content = ""
    try:
        request_options = {
            "headers": { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" }
        }
        page = requests.get(
            url,
            **request_options
        )
        soup = BeautifulSoup(page.content, "html.parser")
        # find text in p, list, pre (github code), td
        chunks = soup.find_all(["h1", "h2", "h3", "h4", "h5", "p", "pre", "td"])
        for chunk in chunks:
            if chunk.name.startswith("h"):
                content += "#" * int(chunk.name[-1]) + " " + chunk.get_text() + "\n"
            else:
                text = chunk.get_text()
                if text and text != "":
                    content += text + "\n"
        # find text in div that class=content
        divs = soup.find("div", class_="content")
        if divs:
            chunks_with_text = divs.find_all(text=True)
            for chunk in chunks_with_text:
                if isinstance(chunk, str) and chunk.strip():
                    content += chunk.strip() + "\n"
        content = re.sub(r"\n+", "\n", content)
        return content
    except Exception as e:
        return f"Can not browse '{ url }'.\tError: { str(e) }"

自定义决策工作流

import nest_asyncio
nest_asyncio.apply()
import asyncio
import Agently
import json

tool_using_workflow = Agently.Workflow()

@tool_using_workflow.chunk()
def save_user_input(inputs, storage):
    storage.set("user_input", inputs["default"])
    return

@tool_using_workflow.chunk()
def make_next_plan(inputs, storage):
    agent = storage.get("$agent")
    user_input = storage.get("user_input")
    tools_info = storage.get("tools_info", {})
    done_plans = storage.get("done_plans", [])
    tools_list = []
    for key, value in tools_info.items():
        tools_list.append({
            "工具名称": key,
            "工具描述": value["desc"],
            "所需参数": value["kwargs"],
        })
    result = (
        agent
            .input(user_input)
            .info({
                "可用工具清单": tools_list,
                "已经做过": done_plans,
            })
            .instruct([
                "根据{input}的用户意图，{已经做过}提供的行动记录以及{可用工具清单}提供的工具，制定解决问题的下一步计划",
                "如果{已经做过}提供的行动记录中，某项行动反复出现错误，可将下一步计划定为'输出结果'，回复内容为对错误的说明",
            ])
            .output({
                "next_step_thinking": ("str", ),
                "next_step_action": {
                    "type": ("'工具使用' | '输出结果'", "MUST IN values provided."),
                    "reply": ("str", "if {next_step_action.type} == '输出结果'，输出你的最终回复结果，else输出''"),                    
                    "tool_using": (
                        {
                            "tool_name": ("str from {可用工具清单.工具名称}", "必须使用{可用工具清单}提供的工具"),
                            "purpose": ("str", "描述使用工具希望解决的问题"),
                            "kwargs": ("dict，根据{可用工具清单.所需参数}要求给出所需参数"),
                        },
                        "if {next_step_action.type} == '工具使用'，给出你的工具使用计划说明，else输出null",
                    ),
                }
            })
            .start()
    )
    return result["next_step_action"]

@tool_using_workflow.chunk()
def reply(inputs, storage):
    if storage.get("print_process"):
        print("[💬 我觉得可以回复了]：")
        print("✅ 我得到的最终结果是：", inputs["default"]["reply"])
    return {
        "reply": inputs["default"]["reply"],
        "process_results": storage.get("done_plans"),
    }

@tool_using_workflow.chunk()
async def use_tool(inputs, storage):
    tool_using_info = inputs["default"]["tool_using"]
    tools_info = storage.get("tools_info")
    tool_func = tools_info[tool_using_info["tool_name"]]["func"]
    if storage.get("print_process"):
        print("[🪛 我觉得需要使用工具]：")
        print("🤔 我想要解决的问题是：", tool_using_info["purpose"])
        print("🤔 我想要使用的工具是：", tool_using_info["tool_name"])
    if asyncio.iscoroutine(tool_func):
        tool_result = await tool_func(**tool_using_info["kwargs"])
    else:
        tool_result = tool_func(**tool_using_info["kwargs"])
    if storage.get("print_process"):
        print("🎉 我得到的结果是：", tool_result[:100], "...")
    done_plans = storage.get("done_plans", [])
    done_plans.append({
        "purpose": tool_using_info["purpose"],
        "tool_name": tool_using_info["tool_name"],
        "result": tool_result,
    })
    storage.set("done_plans", done_plans)
    return

(
    tool_using_workflow
        .connect_to("save_user_input")
        .connect_to("make_next_plan")
        .if_condition(lambda return_value, storage: return_value["type"] == "输出结果")
            .connect_to("reply")
            .connect_to("end")
        .else_condition()
            .connect_to("use_tool")
            .connect_to("make_next_plan")
)

<Agently.Workflow.Chunk.SchemaChunk at 0x115f6f130>

创建Agent并将决策工作流附着（Attach）到Agent上

search_agent = (
    Agently.create_agent()
        .set_settings("current_model", "OAIClient")
        .set_settings("model.OAIClient.url", ENV.DEEPSEEK_URL)
        .set_settings("model.OAIClient.auth", { "api_key": ENV.DEEPSEEK_API_KEY })
        .set_settings("model.OAIClient.options", { "model": ENV.DEEPSEEK_MODEL })
)
search_agent.attach_workflow("tool_using", tool_using_workflow)

<Agently.Agent.Agent.Agent at 0x115f17460>

使用Agent实例方法运行新附着的工作流

result = search_agent.tool_using(
    "Agently框架的Instant模式是什么？能不能给一段代码样例",
    tools_info={
        "search": {
            "desc": "使用网络搜索工具，搜索{keywords}指定关键词相关信息",
            "kwargs": {
                "keywords": [("str", "一个关键词")],
            },
            "func": search,
        },
        "browse": {
            "desc": "使用浏览工具，浏览{url}指定的页面内容",
            "kwargs": {
                "url": ("str", "可访问的URL地址")
            },
            "func": browse,
        },
    },
    print_process=True,
)
print("最终结果:\n", result["default"]["reply"])

[🪛 我觉得需要使用工具]：
🤔 我想要解决的问题是： 搜索Agently框架Instant模式的相关信息
🤔 我想要使用的工具是： search
🎉 我得到的结果是： {"searchParameters":{"q":"Agently Instant模式","type":"search","engine":"google"},"organic":[{"title": ...
[🪛 我觉得需要使用工具]：
🤔 我想要解决的问题是： 浏览Agently框架Instant模式的详细文档以获取代码样例
🤔 我想要使用的工具是： browse
🎉 我得到的结果是： # Agently Instant模型返回结果流式解析
Agently Instant 是3.4.0版本引入的一项重要功能，为什么它的加入值得更新一个第二位大版本号？下面进行具体的介绍说明：
Note ...
[💬 我觉得可以回复了]：
✅ 我得到的最终结果是： 以下是Agently框架Instant模式的代码样例：

```python
import datetime
import Agently

agent = (
    Agently.create_agent()
        #.set_setting(...)
)

# 使用监听器监听新引入的instant事件
@agent.on_event("instant")
def instant_handler(data):
    # 返回的事件数据结构：
    # `key`: <str> 当前正在输出的键（采用Agently Instant表达方法）
    # `indexes`: <list> 如果当前正在输出的键路径中存在数组，`indexes`里会提供当前输出
    #                   是路径中数组的第几项
    # `delta`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新每次添加的新内容
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`value`字段值一致
    # `value`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新当前已生成的全量值
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`delta`字段值一致
    # `complete_value`: <any> 在当前事件抛出时，已经输出的结构化数据的全量内容
    # 输出Instant模式过程结果和输出时间
    print(datetime.now(), data["key"], data["indexes"], data["delta"])

result = (
    agent
        # 使用.use_instant()开启instant模式
        # 3.4.0.3版本之后可以省去此步
        .use_instant()
        .input("Generate 3 other words, then use those 3 words to make a sentence, then generate 4 numbers.")
        # 使用Agently Output语法定义一个复杂结构数据
        .output({
            "words": [("str", )],
            "sentence": ("str", ),
            "numbers": [{ "value": ("int", ) }]
        })
        .start()
)

# 输出最终结果和完成时间
print(datetime.now(), result)
```

运行结果：
Instant模式输出：
2024-11-03 02:20:01.650752 words.[].$delta [0] cat
2024-11-03 02:20:01.831325 words.[].$delta [1] mouse
2024-11-03 02:20:01.835427 words.[] [0] cat
2024-11-03 02:20:01.849140 words.[].$delta [2] run
2024-11-03 02:20:01.850624 words.[] [1] mouse
2024-11-03 02:20:01.912867 words [] ['cat', 'mouse', 'run']
2024-11-03 02:20:01.913157 words.[] [2] run
2024-11-03 02:20:01.962901 sentence.$delta [] The
2024-11-03 02:20:01.980559 sentence.$delta []  cat
2024-11-03 02:20:01.998184 sentence.$delta []  chased
2024-11-03 02:20:02.015376 sentence.$delta []  the
2024-11-03 02:20:02.032466 sentence.$delta []  mouse
2024-11-03 02:20:02.050336 sentence.$delta []  as
2024-11-03 02:20:02.088583 sentence.$delta []  it
2024-11-03 02:20:02.091482 sentence.$delta []  ran
2024-11-03 02:20:02.102013 sentence.$delta []  for
2024-11-03 02:20:02.118886 sentence.$delta []  its
2024-11-03 02:20:02.136612 sentence.$delta []  life
2024-11-03 02:20:02.154099 sentence.$delta [] .
2024-11-03 02:20:02.258635 sentence [] The cat chased the mouse as it ran for its life.
2024-11-03 02:20:02.556008 numbers.[] [0] {'value': 123}
2024-11-03 02:20:02.556662 numbers.[].value [0] 123
2024-11-03 02:20:02.747380 numbers.[] [1] {'value': 456}
2024-11-03 02:20:02.748144 numbers.[].value [1] 456
2024-11-03 02:20:02.938182 numbers.[] [2] {'value': 789}
2024-11-03 02:20:02.938688 numbers.[].value [2] 789
2024-11-03 02:20:03.483925  [] {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}
2024-11-03 02:20:03.484688 numbers [] [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]
2024-11-03 02:20:03.485579 numbers.[] [3] {'value': 101112}
2024-11-03 02:20:03.486465 numbers.[].value [3] 101112
最终Result：
2024-11-03 02:20:03.490869 {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}
```
最终结果:
 以下是Agently框架Instant模式的代码样例：

```python
import datetime
import Agently

agent = (
    Agently.create_agent()
        #.set_setting(...)
)

# 使用监听器监听新引入的instant事件
@agent.on_event("instant")
def instant_handler(data):
    # 返回的事件数据结构：
    # `key`: <str> 当前正在输出的键（采用Agently Instant表达方法）
    # `indexes`: <list> 如果当前正在输出的键路径中存在数组，`indexes`里会提供当前输出
    #                   是路径中数组的第几项
    # `delta`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新每次添加的新内容
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`value`字段值一致
    # `value`: <any> 当前正在输出的键值，如果键值类型是str，此字段更新当前已生成的全量值
    #                否则只在键值完全生成完毕后抛出事件，此时字段值和`delta`字段值一致
    # `complete_value`: <any> 在当前事件抛出时，已经输出的结构化数据的全量内容
    # 输出Instant模式过程结果和输出时间
    print(datetime.now(), data["key"], data["indexes"], data["delta"])

result = (
    agent
        # 使用.use_instant()开启instant模式
        # 3.4.0.3版本之后可以省去此步
        .use_instant()
        .input("Generate 3 other words, then use those 3 words to make a sentence, then generate 4 numbers.")
        # 使用Agently Output语法定义一个复杂结构数据
        .output({
            "words": [("str", )],
            "sentence": ("str", ),
            "numbers": [{ "value": ("int", ) }]
        })
        .start()
)

# 输出最终结果和完成时间
print(datetime.now(), result)
```

运行结果：
Instant模式输出：
2024-11-03 02:20:01.650752 words.[].$delta [0] cat
2024-11-03 02:20:01.831325 words.[].$delta [1] mouse
2024-11-03 02:20:01.835427 words.[] [0] cat
2024-11-03 02:20:01.849140 words.[].$delta [2] run
2024-11-03 02:20:01.850624 words.[] [1] mouse
2024-11-03 02:20:01.912867 words [] ['cat', 'mouse', 'run']
2024-11-03 02:20:01.913157 words.[] [2] run
2024-11-03 02:20:01.962901 sentence.$delta [] The
2024-11-03 02:20:01.980559 sentence.$delta []  cat
2024-11-03 02:20:01.998184 sentence.$delta []  chased
2024-11-03 02:20:02.015376 sentence.$delta []  the
2024-11-03 02:20:02.032466 sentence.$delta []  mouse
2024-11-03 02:20:02.050336 sentence.$delta []  as
2024-11-03 02:20:02.088583 sentence.$delta []  it
2024-11-03 02:20:02.091482 sentence.$delta []  ran
2024-11-03 02:20:02.102013 sentence.$delta []  for
2024-11-03 02:20:02.118886 sentence.$delta []  its
2024-11-03 02:20:02.136612 sentence.$delta []  life
2024-11-03 02:20:02.154099 sentence.$delta [] .
2024-11-03 02:20:02.258635 sentence [] The cat chased the mouse as it ran for its life.
2024-11-03 02:20:02.556008 numbers.[] [0] {'value': 123}
2024-11-03 02:20:02.556662 numbers.[].value [0] 123
2024-11-03 02:20:02.747380 numbers.[] [1] {'value': 456}
2024-11-03 02:20:02.748144 numbers.[].value [1] 456
2024-11-03 02:20:02.938182 numbers.[] [2] {'value': 789}
2024-11-03 02:20:02.938688 numbers.[].value [2] 789
2024-11-03 02:20:03.483925  [] {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}
2024-11-03 02:20:03.484688 numbers [] [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]
2024-11-03 02:20:03.485579 numbers.[] [3] {'value': 101112}
2024-11-03 02:20:03.486465 numbers.[].value [3] 101112
最终Result：
2024-11-03 02:20:03.490869 {'words': ['cat', 'mouse', 'run'], 'sentence': 'The cat chased the mouse as it ran for its life.', 'numbers': [{'value': 123}, {'value': 456}, {'value': 789}, {'value': 101112}]}
```

python print("处理过程:\n", json.dumps(result["default"]["process_results"], indent=4, ensure_ascii=False))

十、做个总结

• Agently AnyModel：模型随心换，不用修改业务代码

  点击此处查看AnyModel文档

• Agently AgenticRequest：为工程开发设计的运行时模型请求代理（Agent），结构化输出控制、插件式能力增强，统统简单搞定

  点击此处查看AgenticRequest入门

• Agently Workflow：清晰表达的心流体验，文思泉涌，帮助你轻松编排你的AI应用

  点击此处查看Workflow文档

• Agently Instant模式：结构化输出和流式输出再也不用二选一，快人一步，灵活高效

  点击此处查看Agently Instant模式介绍及实验说明

• Agently P-YAML：通过YAML文件将模型请求方法管理从代码中分离，大模型应用时代的新协同范式

  点击此处查看Agently P-YAML方法使用说明

欢迎扫码加入微信群聊继续讨论：