python机器人Agent编程——使用swarm框架和ollama实现一个本地大模型和爬虫结合的手机号归属地天气查询Agent流（体会）

🕗 发布于 2024-11-10 09:23 python 机器人爬虫

一、前言

前一篇博文，我们使用qwen_agent库，及fastapi服务来实现了一个手机号归属地天气查询工具，运行发现一个问题，就是大部分运行时，系统时无法实现多个agent的调用的，很多情况下一次查询只能调用一个function。搜索了一些相关资料，发现，目前很多多agent框架也没有很好的解决这样的问题。个人判断，目前这项技术应该还没有成熟，或者开源的框架还没有很好的表现。比如crewai等。最佳openai开源了swarm框架，试了一下，也没找到一个很好的办法。本篇，实现的是固定的流程，就是手动设置几个agent处理消息的流程，主要是串行的流程，来实现输入手机号码，三个上下游的agent像流水线一样，进行相应的信息处理包括调用function，最后返回城市及天气信息：
在这里插入图片描述

二、代码实现

2.1关于本地模型设置

这里需要注意的是，要实现swarm调用ollama本地模型，本人测试，大模型版本必须是qwen需要是2.5，和llama必须是3.2，低于这个版本，会报不能调用tool功能的错误。swarm默认是openai的自家的模型，因此这里会存在兼容性的问题，如果用到生产环境，不排除可能会有硬伤。
要实现ollama本地模型，参考了网上的办法，需要做如下设置：

  
from openai import OpenAI
# 定义本地模型  
MODEL = "llama3.2:latest"  
#MODEL = "qwen2.5:0.5b"
ollama_client = OpenAI(
    base_url = 'http://localhost:11434/v1',
    api_key='None', # required, but unused
)
# 初始化 Swarm 客户端用本地模型 
client = Swarm(client=ollama_client)

2.2智能体的定义

智能体定义比较简洁，这个非常直观，需要注意的是，如果是使用本地大模型，每个agnet定义时，都要指定模型的名称，例如：


  
# 创建1个智能体  
search_agent = Agent(  
    name="receptionist agent",  
    instructions="""  
    You are a receptionist, and you need to respond according to the user's requests. Your tasks are: 
    1) If the user provides a phone number and does not ask about the weather, you should call the get_mobile_address function and then directly return the query result; 
    2) If the user provides a phone number and requests to check the weather, you should first call the mobile phone origin query tool, then call the weather query tool, and finally return the query results.

    """,  
    functions=[get_mobile_address],  #这里定义这个智能体可以使用的function
    model=MODEL  #这里需要指定前面定义的模型
)

2.3定义调用的function

智能体是可以调用function（也叫tool？）的，swarm的function定义相较于qwen-agent和lanchain等框架，现得非常轻量，就是常规的function格式就ok，他会主动帮你生成json schema。

# 定义手机号归属地搜索 Function，并返回查询结果  
def get_mobile_address(number:str):  
    """Search for the location according the number input"""     
    res=p.find(number)
    print("1原始查询结果:",type(res),res)
    if res:
        if "city" in res.keys(): 
            key=  res['city']    
            if key in citycode:                
                code=citycode[key]       
                print("城市的代码：",code)
                return f"according the phon number:{number},it belong to the city whose name is:{key},according the name of city I find the citynumber,the citynumber is "+code
            else:
                print("未找到城市的代码")
                return "sorry I cant find the citynumber"            
            
        else:
            print("未找到手机号归属地，抱歉")
            return "sorry I cant find the citynumber"
    else:
        print("未找到手机号归属地，抱歉")
        return "未找到手机号归属地，抱歉"

2.4定义Agent的流

这里可以设置一个Agent先后调用的串行流程，本例流程是用户输入了一个电话号码，接下来第一个agent拿到用户的输入，就根据用户的输入是否含有电话号码，如果有就调用一个funciton查归属地，如果没有就不调用（这里发现无论输没输电话号码都会调用function,估计跟大模型的智商有关系，估计好的模型应该不会这么傻叉，待验证)。第一个Agent完成任务后，就交给第二个Agent，第二个Agent根据第一个完成情况，进行查询天气或者直接返回无法查询，然后交给第三个Agent，这个Agent负责接收第二个的结果信息，进行总结提炼，然后形成一个报告返回给用户，流程结束。

# 工作流程，按顺序处理
def process_ask(topic):  
    """Run the searching workflow"""  
    with st.spinner("Processing ..."):  # Changed to st.spinner for better context
        # Search  
        st.write(f"1.Searching the location for your phone number...{topic}")  
        search_response = client.run(  
            agent=search_agent,  
            messages=[{"role": "user", "content": f"what is location of this number:{topic}"}]  
        )  
        response1 = search_response.messages[-2]["content"] 
        print("search_response.messages\n",response1)
        # Synthesize  
        st.write(f"2.weather_agent processing...")  
        synthesis_response = client.run(  
            agent=weather_agent,  
            messages=[{"role": "user", "content": f"{response1}"}]  
        )  
        response2 = synthesis_response.messages[-1]["content"]  
        print("synthesis_response.messages\n",response2)
        # Summarize  
        st.write(f"3.Creating summary...")  
        summary_response = client.run(  
            agent=summary_agent,  
            messages=[{"role": "user", "content": f"Summarize this information:\n{response2}"}]  
        )  
     
        return summary_response.messages[-1]["content"]

2.5发布到streamlit的交互页上

加上一层web客户端，用streamlit框架，接收用户输入，返回流程进度及查询结果信息：

# 用户web交互界面  
topic = st.text_input("Enter your number:", value="13858130853")  
if st.button("Process searching", type="primary"):  
    if topic:  
        try:  
            final_summary = process_ask(topic)  
            st.header(f"Searching: {topic}")  
            st.markdown(final_summary)  
        except Exception as e:  
            st.error(f"An error occurred: {str(e)}")  
    else:  
        st.error("Please enter a number!")

三、完整代码

完整代码如下：

import streamlit as st  
from swarm import Swarm, Agent  

import requests
import re
from phone import Phone
from city import *
  
from openai import OpenAI
# 定义模型  
MODEL = "llama3.2:latest"  
#MODEL = "qwen2.5:0.5b"
ollama_client = OpenAI(
    base_url = 'http://localhost:11434/v1',
    api_key='None', # required, but unused
)
# 初始化 Swarm 客户端  
client = Swarm(client=ollama_client)  
  
# 通过 Streamlit 创建用户界面，为页面和应用程序添加一个标题  
st.set_page_config(page_title="AI News Processor", page_icon="📰")  
st.title("📰 Phone Location And Weather Searcher Agent")  

p = Phone()  
# 定义新闻搜索 Function，使用 DuckDuckGo 搜索 API，获取当月新闻，并返回结构化结果  
def get_mobile_address(number:str):  
    """Search for the location according the number input"""     
    res=p.find(number)
    print("1原始查询结果:",type(res),res)
    if res:
        if "city" in res.keys(): 
            key=  res['city']    
            if key in citycode:                
                code=citycode[key]       
                print("城市的代码：",code)
                return f"according the phon number:{number},it belong to the city whose name is:{key},according the name of city I find the citynumber,the citynumber is "+code
            else:
                print("未找到城市的代码")
                return "sorry I cant find the citynumber"            
            
        else:
            print("未找到手机号归属地，抱歉")
            return "sorry I cant find the citynumber"
    else:
        print("未找到手机号归属地，抱歉")
        return "未找到手机号归属地，抱歉"


def get_weather(citynumber: str):
    """Search for the weather according the city name"""     
    print("2 输入查询的天气对应城市号码：",citynumber)
    try:
        url=f'https://www.weather.com.cn/weather1d/{citynumber}.shtml#input'         
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0",
            "Cookie": "userNewsPort0=1; defaultCty=101010100; defaultCtyName=%u5317%u4EAC; f_city=%E5%8C%97%E4%BA%AC%7C101010100%7C; Hm_lvt_080dabacb001ad3dc8b9b9049b36d43b=1730788105,1730874362; HMACCOUNT=5E07A605B65ACF5F; Hm_lpvt_080dabacb001ad3dc8b9b9049b36d43b=1730875029"
        }
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            html_content = response.content.decode('utf-8')  # Decode response content to support Chinese
            #print(html_content)
            # 使用正则表达式查找<script>标签中的变量
            script_matches = re.findall(r'<script.*?>(.*?)</script>', html_content, re.DOTALL)
            match=False
            if len(script_matches)>4:
                hour3data = script_matches[4] 
                import json
                
                # Extract the hour3data variable from the script
                hour3data_str = hour3data
                
                # Parse the string to extract the 7d data
                hour3data_json = json.loads(hour3data_str.split('=')[1].strip().rstrip(';'))
                seven_day_data = hour3data_json["7d"]
                
                # Format the extracted data for output
                formatted_data = []
                for day in seven_day_data:
                    formatted_data.append(", ".join(day))
                
                # Join the formatted data into a single string
                hour3data_output = "\n".join(formatted_data)
                print("提取的7天天气数据:", hour3data_output)  
                from datetime import datetime
                # Get the current date and time
                current_datetime = datetime.now()
                current_date_hour = current_datetime.strftime("%Y年-%m月-%d日 %H:00时")
                print("当前日期到小时:", current_date_hour)
                return f"今天的日期和小时时间是 {current_date_hour},the next hours of today's weather and the next 7 day's weather datas is follows:"+hour3data
            else:
                print("未找到天气小时数据")
                return "未找到天气小时数据"
        else:
            print("请求失败，状态码")
            return "请求失败，状态码: {}".format(response.status_code)        
      
    except : 
        print("error")       
        return "error"


  
# 创建智能体  
search_agent = Agent(  
    name="receptionist agent",  
    instructions="""  
    You are a receptionist, and you need to respond according to the user's requests. Your tasks are: 
    1) If the user provides a phone number and does not ask about the weather, you should call the get_mobile_address function and then directly return the query result; 
    2) If the user provides a phone number and requests to check the weather, you should first call the mobile phone origin query tool, then call the weather query tool, and finally return the query results.

    """,  
    functions=[get_mobile_address],  
    model=MODEL  
)  


weather_agent = Agent(  
    name="weather search agent",  
    instructions="""  
    you are a weather search server,your task are:
    1) if the user find a citynumber (which begin with "10" and must be a number), you should use the value of citynumbe(which begin with "10" and must be a number) as parameter and call the get_weather funciton to get the weather information.
    2) if the user did not find a citynumber,do not call any funciton and just respone the orignal message.
    """,  
    functions=[get_weather],  
    model=MODEL  
)  
  
summary_agent = Agent(  
    name="Summarizer weather reporter agent",  
    instructions="""  
    You are a summarization weather reporter, who can analyzes the input information. 
    If there is weather information, Based on the data from the next few hours and the data for the next 7 days, you summarize the weather for today and the weather trends for the next 7 days , and give a weather report,
    you should strictly using the information and do not say anything else out of the information, at the head of the report,you should say the city name belong to phone number,speak in simple chinese.
.   If there is no weather information, you should say sorry to the user.
    """,      
    model=MODEL  
)  
  
# 实施新闻处理工作流程，按顺序处理，显示进度指标  
def process_ask(topic):  
    """Run the searching workflow"""  
    with st.spinner("Processing ..."):  # Changed to st.spinner for better context
        # Search  
        st.write(f"1.Searching the location for your phone number...{topic}")  
        search_response = client.run(  
            agent=search_agent,  
            messages=[{"role": "user", "content": f"what is location of this number:{topic}"}]  
        )  
        response1 = search_response.messages[-2]["content"] 
        print("search_response.messages\n",response1)
        # Synthesize  
        st.write(f"2.weather_agent processing...")  
        synthesis_response = client.run(  
            agent=weather_agent,  
            messages=[{"role": "user", "content": f"{response1}"}]  
        )  
        response2 = synthesis_response.messages[-1]["content"]  
        print("synthesis_response.messages\n",response2)
        # Summarize  
        st.write(f"3.Creating summary...")  
        summary_response = client.run(  
            agent=summary_agent,  
            messages=[{"role": "user", "content": f"Summarize this information:\n{response2}"}]  
        )  
     
        return summary_response.messages[-1]["content"]  
  
# 用户web交互界面  
topic = st.text_input("Enter your number:", value="13858130853")  
if st.button("Process searching", type="primary"):  
    if topic:  
        try:  
            final_summary = process_ask(topic)  
            st.header(f"Searching: {topic}")  
            st.markdown(final_summary)  
        except Exception as e:  
            st.error(f"An error occurred: {str(e)}")  
    else:  
        st.error("Please enter a number!")

四、总结

经过来回运行调试中发现，有如下体会：

本地大模型质量对于Agent的表现非常关键，同样的llama3.2:latest比qwen2.5:0.5b要表现好
提示词也很关键，要写好instructions，这个需要下功夫，反复调试，最好系统学习一下提示工程
中英文混用，可能会影响模型的表现
总体来说多智能体，开源出来的框架表现应该都不佳，没有真正实现模型的自动流程编排，这可能是某些公司的核心技术，还没有公布透露？网友谁有研究请不吝留言赐教。
多智能体协作确实是未来的方向，希望能够出来更智能的框架和很好支持多智能体的本地大模型。
多智能体是解决具身智能自主决策的解决方案

亲爱的网友，您怎么看？

[------------本篇完-------------]

PS.扩展阅读

————————————————————————————————————————

对于python机器人编程感兴趣的小伙伴，可以进入如下链接阅读相关咨询

ps1.六自由度机器人相关文章资源

(1) 对六自由度机械臂的运动控制及python实现（附源码)
在这里插入图片描述

(2) N轴机械臂的MDH正向建模，及python算法
在这里插入图片描述

在这里插入图片描述

（2）python机器人实战——0到1创建一个自动是色块机器人项目-CSDN直播

(3)博文《我从0开始搭建了一个色块自动抓取机器人，并实现了大模型的接入和语音控制-(上基础篇)》的vrep基础环境
(3)博文《我从0开始搭建了一个色块自动抓取机器人，并实现了大模型的接入和语音控制-(上基础篇)》的vrep基础环境
(4)实现了语音输入+大模型指令解析+机器视觉+机械臂流程打通
在这里插入图片描述

(3)python机器人编程——差速AGV机器、基于视觉和预测控制的循迹、自动行驶（上篇）_agv编程-CSDN博客
(4)python机器人编程——差速AGV机器、基于视觉和预测控制的循迹、自动行驶（下篇）_agv路线规划原则python-CSDN博客
对应python及仿真环境资源：源码链接
在这里插入图片描述

ps3.wifi小车控制相关文章资源

web端配套资源源代码已经上传(竖屏版)，下载地址
仿真配套资源已经上传：下载地址
web端配套资源源代码已经上传(横屏版)，下载地址

原文地址：https://blog.csdn.net/kanbide/article/details/143652534

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：单相锁相环，原理与Matlab实现
下一篇：SSH实验1

第二十六周机器学习笔记：PINN求正反解求PDE文献阅读——正问题
本周周报主要详细分析了一篇PINN求正反解的经典文献，文献主要讲述了作者是如何利用PINN正反解求PDE的。之前的深度神经网络在缺乏数据的情况下很容易发生过拟合现象且缺乏鲁棒性。相比较之下，PINN的
阅读更多2024-12-27
机器学习基础衡量模型性能指标
机器学习中的一些常见性能评估指标
阅读更多2024-12-27
Jmeter自学【8】- 使用JMeter模拟设备通过MQTT发送数据
今天使用jmeter推送数据到MQTT，给大家分享一下操作流程。
阅读更多2024-12-27
Idea2024找不到Ant Build
平常使用的是Idea2024版本，虽然不常用Ant Build，但是用到的时候发现找不到这个功能，原来是Idea2024默认没有添加这个插件。
阅读更多2024-12-27
C++简明教程（4）（Hello World）
想象一下，在现实世界中，可能有两个不同的城市都有一个叫做“张庄”的地方，如果我们在地图上只写“张庄”，就很难区分到底是指哪个城市的“张庄”。就好像你在写一篇故事，故事里的角色要使用某个特定的工具，但这
阅读更多2024-12-27
upload-labs关卡记录2
同理，这里依旧进行上传我们的一句话木马进行测试，这里可以看到，页面进行重新加载了，因此我们判断不是客户端检测，而是服务端检测。这里就尝试对content-type进行更改，改为图片格式，尝试进行绕过。
阅读更多2024-12-27
【NIFI】实现MySQL-＞ORACLE数据同步
实现MySQL->ORACLE数据同步。
阅读更多2024-12-27
一个基于Rust适用于 Web、桌面、移动设备等的全栈应用程序框架
Dioxus项目的诞生源于开发者们对于更高效、更灵活的跨平台UI解决方案的渴望。随着技术的发展，用户对于应用的需求不再局限于单一的操作系统或设备类型，而是希望能够在不同的平台上获得无缝衔接的体验。然而
阅读更多2024-12-27
【网络】超以太网联盟 UEC|下一代 “RoCE” 协议--编辑中
Tail latency: 尾部延迟，（以通信阶段最后一条消息的到达时间为衡量标准）是系统性能的关键指标。UET: 超级以太传输协议。UEC: 超级以太联盟。
阅读更多2024-12-27
如何通过运行时威胁洞察提升反欺诈策略
当威胁数据表明某一特定安全漏洞或攻击模式在多个机构之间蔓延时，金融机构可以联合同行，分享发现的威胁模式和应对策略，从而更有效地识别和遏制新型威胁的传播。借助先进的威胁检测与响应技术，企业不仅能够应对当
阅读更多2024-12-27