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

一、前言

前一篇博文，我们使用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!")  

四、总结

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

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

亲爱的网友，您怎么看？

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

PS.扩展阅读

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

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

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

(1) 对六自由度机械臂的运动控制及python实现（附源码)

(2) N轴机械臂的MDH正向建模，及python算法

ps2.四轴机器相关文章资源

(1) 文章：python机器人编程——用python实现一个写字机器人

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

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

ps3.移动小车相关文章资源

（1）python做了一个极简的栅格地图行走机器人，到底能干啥？[第五弹]——解锁蒙特卡洛定位功能-CSDN博客
(2) 对应python资源：源码地址

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

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

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